diff --git a/core/crates/solstone-core-sol-client/src/error.rs b/core/crates/solstone-core-sol-client/src/error.rs index 97708f12c..7f2398933 100644 --- a/core/crates/solstone-core-sol-client/src/error.rs +++ b/core/crates/solstone-core-sol-client/src/error.rs @@ -90,6 +90,14 @@ impl ClientError { ClientError::Unreachable { .. } | ClientError::Timeout { .. } => None, } } + + #[must_use] + pub fn payload(&self) -> Option<&serde_json::Value> { + match self { + ClientError::ReasonRejected { payload, .. } => Some(payload), + _ => None, + } + } } impl fmt::Display for ClientError { diff --git a/core/fixtures/native-sol/parity/link.jsonl b/core/fixtures/native-sol/parity/link.jsonl index 9d94504ab..350990b97 100644 --- a/core/fixtures/native-sol/parity/link.jsonl +++ b/core/fixtures/native-sol/parity/link.jsonl @@ -6,8 +6,8 @@ {"id":"link-private-link-status-service-down","surface":"sol-call","argv":["link","private-link","status"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"GET","path":"/app/network/api/private-link","fault":{"kind":"service_down"}}]},"normalizations":[],"expected":{"stdout":"","stderr":"sol: solstone isn't running. Start it with 'journal up' and retry.\n","exit":1,"requests":[{"method":"GET","path":"/app/network/api/private-link","query":[],"json":null,"headers":[],"timeout_policy":"api"}]}} {"id":"link-private-link-disable-success","surface":"sol-call","argv":["link","private-link","disable"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/private-link/disable","response":{"status":200,"json":{"success":true,"service":"spl","result":{"was_enabled":true},"status":{"success":true,"service":"spl","state":"not_enabled","posture":"direct","enrolled":false,"relay_url":"https://relay.example","actions":{"enable":true,"disable":false},"operation":null}}}}]},"normalizations":[],"expected":{"stdout":"your private network is off. devices connect directly again.\n","stderr":"","exit":0,"requests":[{"method":"POST","path":"/app/network/private-link/disable","query":[],"json":null,"headers":[],"timeout_policy":"api"}]}} {"id":"link-private-link-disable-failed","surface":"sol-call","argv":["link","private-link","disable"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/private-link/disable","fault":{"kind":"reason","status":500,"reason_code":"service_operation_failed","error":"service_operation_failed"}}]},"normalizations":[],"expected":{"stdout":"","stderr":"couldn't turn off your private network — it's still on. try again.\n","exit":1,"requests":[{"method":"POST","path":"/app/network/private-link/disable","query":[],"json":null,"headers":[],"timeout_policy":"api"}]}} -{"id":"link-unpair-label-success","surface":"sol-call","argv":["link","unpair","Phone"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/unpair","json":{"device_label":"Phone"},"response":{"status":200,"json":{"unpaired":"sha256:aaa111"}}}]},"normalizations":[],"expected":{"stdout":"Unpaired.\n","stderr":"","exit":0,"requests":[{"method":"POST","path":"/app/network/unpair","query":[],"json":{"device_label":"Phone"},"headers":[],"timeout_policy":"api"}]}} -{"id":"link-unpair-fingerprint-success","surface":"sol-call","argv":["link","unpair","sha256:aaa111"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/unpair","json":{"fingerprint":"sha256:aaa111"},"response":{"status":200,"json":{"unpaired":"sha256:aaa111"}}}]},"normalizations":[],"expected":{"stdout":"Unpaired.\n","stderr":"","exit":0,"requests":[{"method":"POST","path":"/app/network/unpair","query":[],"json":{"fingerprint":"sha256:aaa111"},"headers":[],"timeout_policy":"api"}]}} +{"id":"link-unpair-label-success","surface":"sol-call","argv":["link","unpair","Phone"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/unpair","json":{"device_label":"Phone"},"response":{"status":200,"json":{"unpaired":"sha256:aaa111","revoked_observers":[{"name":"phone-a","prefix":"phone-a-"},{"name":"phone-b","prefix":"phone-b-"}]}}}]},"normalizations":[],"expected":{"stdout":"Unpaired.\nRevoked observer streams: phone-a, phone-b\n","stderr":"","exit":0,"requests":[{"method":"POST","path":"/app/network/unpair","query":[],"json":{"device_label":"Phone"},"headers":[],"timeout_policy":"api"}]}} +{"id":"link-unpair-fingerprint-success","surface":"sol-call","argv":["link","unpair","sha256:aaa111"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/unpair","json":{"fingerprint":"sha256:aaa111"},"response":{"status":200,"json":{"unpaired":"sha256:aaa111","revoked_observers":[]}}}]},"normalizations":[],"expected":{"stdout":"Unpaired.\nNo bound observer streams were revoked.\n","stderr":"","exit":0,"requests":[{"method":"POST","path":"/app/network/unpair","query":[],"json":{"fingerprint":"sha256:aaa111"},"headers":[],"timeout_policy":"api"}]}} {"id":"link-unpair-label-not-found","surface":"sol-call","argv":["link","unpair","Missing"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/unpair","json":{"device_label":"Missing"},"fault":{"kind":"reason","status":404,"reason_code":"paired_device_not_found","error":"not found","detail":"no paired device with that label"}}]},"normalizations":[],"expected":{"stdout":"No paired device with label 'Missing'\n","stderr":"","exit":1,"requests":[{"method":"POST","path":"/app/network/unpair","query":[],"json":{"device_label":"Missing"},"headers":[],"timeout_policy":"api"}]}} {"id":"link-pair-invalid-role","surface":"sol-call","argv":["link","pair","--as","bad"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[]},"normalizations":[],"expected":{"stdout":"","stderr":"invalid role; expected one of: phone, observer, peer\n","exit":2,"requests":[]}} {"id":"link-pair-request-invalid","surface":"sol-call","argv":["link","pair"],"env":{},"stdin":"","files":{},"clock":{"today":"20260723","unix_seconds":3600},"transport":{"requests":[{"method":"POST","path":"/app/network/pair-start","json":{"device_label":""},"fault":{"kind":"reason","status":400,"reason_code":"pairing_request_invalid","error":"invalid","detail":"pair-link requires an IPv4 LAN address; none found"}}]},"normalizations":[],"expected":{"stdout":"","stderr":"can't start pairing — your journal isn't reachable on a network address yet. turn on your private network to pair from anywhere, or connect this device to your home network.\n","exit":1,"requests":[{"method":"POST","path":"/app/network/pair-start","query":[],"json":{"device_label":""},"headers":[],"timeout_policy":"api"}]}} diff --git a/solstone/apps/network/native/command.rs b/solstone/apps/network/native/command.rs index 4c4fd0643..08140ebde 100644 --- a/solstone/apps/network/native/command.rs +++ b/solstone/apps/network/native/command.rs @@ -406,7 +406,7 @@ pub fn unpair(ctx: CommandContext<'_>) -> CommandOutput { vec![], Some(payload), ) { - Ok(_response) => stdout_line("Unpaired."), + Ok(response) => unpair_success_output(&response), Err(error) => { if error.reason_code() == Some("paired_device_not_found") { let message = if target.starts_with("sha256:") { @@ -420,11 +420,73 @@ pub fn unpair(ctx: CommandContext<'_>) -> CommandOutput { exit: 1, }; } + if let Some(output) = unpair_partial_failure_output(&error) { + return output; + } link_error(error) } } } +fn unpair_success_output(response: &Value) -> CommandOutput { + let mut lines = vec!["Unpaired.".to_string()]; + let names = revoked_observer_names(response); + if names.is_empty() { + lines.push("No bound observer streams were revoked.".to_string()); + } else { + lines.push(format!("Revoked observer streams: {}", names.join(", "))); + } + stdout(lines) +} + +fn unpair_partial_failure_output(error: &ClientError) -> Option { + if error.reason_code() != Some("internal_error") { + return None; + } + let payload = error.payload()?; + if payload.get("failed_operation").and_then(Value::as_str) != Some("observer_revoke") { + return None; + } + payload.get("unpaired").and_then(Value::as_str)?; + let mut stdout_value = String::new(); + push_line(&mut stdout_value, "Unpaired."); + let names = revoked_observer_names(payload); + if names.is_empty() { + push_line( + &mut stdout_value, + "No bound observer streams were revoked before failure.", + ); + } else { + push_line( + &mut stdout_value, + format!( + "Revoked observer streams before failure: {}", + names.join(", ") + ), + ); + } + Some(CommandOutput { + stdout: stdout_value, + stderr: "Error: failed to revoke all bound observer streams.\n".to_string(), + exit: 1, + }) +} + +fn revoked_observer_names(response: &Value) -> Vec { + response + .get("revoked_observers") + .and_then(Value::as_array) + .map(|items| { + items + .iter() + .filter_map(|item| item.get("name").and_then(Value::as_str)) + .filter(|name| !name.is_empty()) + .map(str::to_string) + .collect() + }) + .unwrap_or_default() +} + fn devices(ctx: CommandContext<'_>) -> Result, ClientError> { let body = request_json( ctx, diff --git a/solstone/apps/network/routes.py b/solstone/apps/network/routes.py index 22a115fe1..c6e96a557 100644 --- a/solstone/apps/network/routes.py +++ b/solstone/apps/network/routes.py @@ -55,13 +55,17 @@ from solstone.apps.network.relay_link import ( derive_rk, encode_pair_window_link, ) -from solstone.apps.observer.utils import revoke_observers_bound_to_device +from solstone.apps.observer.utils import ( + ObserverRevokeError, + revoke_observers_bound_to_device, +) from solstone.apps.utils import log_app_action from solstone.convey import emit from solstone.convey.bridge import get_cached_state from solstone.convey.reasons import ( CONVEY_OPERATION_FAILED, FILE_READ_FAILED, + INTERNAL_ERROR, INVALID_CONFIG_VALUE, INVALID_OPERATION_FOR_STATE, INVALID_REQUEST_VALUE, @@ -989,6 +993,17 @@ def pair() -> Any: return jsonify(response) +def _revoked_observer_projection(observers: list[dict]) -> list[dict[str, str]]: + projected = [ + { + "name": str(observer.get("name") or ""), + "prefix": str(observer.get("filename_prefix") or ""), + } + for observer in observers + ] + return sorted(projected, key=lambda item: (item["name"], item["prefix"])) + + @network_bp.route("/rename", methods=["POST"]) def rename() -> Any: """Rename a paired device by fingerprint.""" @@ -1092,8 +1107,24 @@ def unpair() -> Any: authorized.remove(fingerprint) else: authorized.remove(fingerprint) - revoke_observers_bound_to_device(fingerprint) - return jsonify({"unpaired": fingerprint}) + try: + revoked_observers = revoke_observers_bound_to_device(fingerprint) + except ObserverRevokeError as exc: + return error_response( + INTERNAL_ERROR, + detail="Failed to revoke one or more bound observer streams.", + extra={ + "unpaired": fingerprint, + "revoked_observers": _revoked_observer_projection(exc.revoked), + "failed_operation": "observer_revoke", + }, + ) + return jsonify( + { + "unpaired": fingerprint, + "revoked_observers": _revoked_observer_projection(revoked_observers), + } + ) def _entry_to_json(entry: ClientEntry) -> dict[str, Any]: diff --git a/solstone/apps/network/tests/test_unpair_record_revoke_parity.py b/solstone/apps/network/tests/test_unpair_record_revoke_parity.py index f8128ee1b..c8fef5afa 100644 --- a/solstone/apps/network/tests/test_unpair_record_revoke_parity.py +++ b/solstone/apps/network/tests/test_unpair_record_revoke_parity.py @@ -8,6 +8,7 @@ import logging from importlib import import_module import solstone.apps.network.routes as link_routes +import solstone.apps.observer.utils as observer_utils from solstone.apps.observer.utils import load_observer, save_observer from solstone.think.link.auth import AuthorizedClients from solstone.think.link.paths import authorized_clients_path @@ -56,10 +57,32 @@ def _save_bound_observer(handle: str, name: str, fingerprint: str) -> None: ) +def _save_unbound_observer(handle: str, name: str) -> None: + assert save_observer( + { + "key": handle, + "name": name, + "created_at": 1, + "enabled": True, + "revoked": False, + "stats": {"segments_received": 0, "bytes_received": 0}, + } + ) + + def _post_unpair(env, payload: dict): return env.client.post("/app/network/unpair", json=payload) +def _unpair_payload( + fingerprint: str, revoked_observers: list[dict] | None = None +) -> dict: + return { + "unpaired": fingerprint, + "revoked_observers": revoked_observers or [], + } + + def _action_entries(env) -> list[dict]: actions_dir = env.journal / "config" / "actions" entries = [] @@ -78,7 +101,7 @@ def test_unpair_phone_by_fingerprint_removes_authorized(link_env) -> None: response = _post_unpair(env, {"fingerprint": PHONE_FINGERPRINT}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": PHONE_FINGERPRINT} + assert response.get_json() == _unpair_payload(PHONE_FINGERPRINT) assert _authorized().is_authorized(PHONE_FINGERPRINT) is False assert load_journal_source_by_fingerprint(PHONE_FINGERPRINT) is None @@ -93,13 +116,65 @@ def test_unpair_phone_revokes_bound_observer_records(link_env) -> None: response = _post_unpair(env, {"fingerprint": PHONE_FINGERPRINT}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": PHONE_FINGERPRINT} + assert response.get_json() == _unpair_payload( + PHONE_FINGERPRINT, + [ + {"name": "phone-a", "prefix": "phone-a-"}, + {"name": "phone-b", "prefix": "phone-b-"}, + ], + ) assert _authorized().is_authorized(PHONE_FINGERPRINT) is False assert load_observer("phone-a-observer")["revoked"] is True assert load_observer("phone-b-observer")["revoked"] is True assert load_observer("other-observer")["revoked"] is False +def test_unpair_phone_leaves_unbound_observer_records_unrevoked(link_env) -> None: + env = link_env() + _add_authorized(PHONE_FINGERPRINT, "phone") + _save_unbound_observer("phone-a-observer", "phone-a") + _save_unbound_observer("phone-b-observer", "phone-b") + + response = _post_unpair(env, {"fingerprint": PHONE_FINGERPRINT}) + + assert response.status_code == 200 + assert response.get_json() == _unpair_payload(PHONE_FINGERPRINT) + assert _authorized().is_authorized(PHONE_FINGERPRINT) is False + assert load_observer("phone-a-observer")["revoked"] is False + assert load_observer("phone-b-observer")["revoked"] is False + + +def test_unpair_partial_observer_revoke_failure_reports_saved_revocations( + link_env, + monkeypatch, +) -> None: + env = link_env() + _add_authorized(PHONE_FINGERPRINT, "phone") + _save_bound_observer("phone-a-observer", "phone-a", PHONE_FINGERPRINT) + _save_bound_observer("phone-b-observer", "phone-b", PHONE_FINGERPRINT) + real_save_observer = observer_utils.save_observer + + def fail_second_bound_observer(observer: dict) -> bool: + if observer.get("name") == "phone-b": + return False + return real_save_observer(observer) + + monkeypatch.setattr(observer_utils, "save_observer", fail_second_bound_observer) + + response = _post_unpair(env, {"fingerprint": PHONE_FINGERPRINT}) + + assert response.status_code == 500 + body = response.get_json() + assert body["reason_code"] == "internal_error" + assert body["detail"] == "Failed to revoke one or more bound observer streams." + assert body["unpaired"] == PHONE_FINGERPRINT + assert body["failed_operation"] == "observer_revoke" + assert body["revoked_observers"] == [{"name": "phone-a", "prefix": "phone-a-"}] + assert _authorized().is_authorized(PHONE_FINGERPRINT) is False + assert load_observer("phone-a-observer")["revoked"] is True + assert load_observer("phone-b-observer")["revoked"] is False + + def test_unpair_cascade_runs_after_authorized_removal( link_env, monkeypatch, @@ -123,7 +198,7 @@ def test_unpair_cascade_runs_after_authorized_removal( response = _post_unpair(env, {"fingerprint": PHONE_FINGERPRINT}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": PHONE_FINGERPRINT} + assert response.get_json() == _unpair_payload(PHONE_FINGERPRINT) assert authorized_present_during_cascade == [False] @@ -138,7 +213,7 @@ def test_unpair_unknown_role_removes_authorized_without_warning( response = _post_unpair(env, {"fingerprint": UNKNOWN_ROLE_FINGERPRINT}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": UNKNOWN_ROLE_FINGERPRINT} + assert response.get_json() == _unpair_payload(UNKNOWN_ROLE_FINGERPRINT) assert _authorized().is_authorized(UNKNOWN_ROLE_FINGERPRINT) is False assert [ record @@ -162,7 +237,7 @@ def test_unpair_peer_revokes_source_removes_authorized_and_logs_action( response = _post_unpair(env, {"device_label": "peer"}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": PEER_FINGERPRINT} + assert response.get_json() == _unpair_payload(PEER_FINGERPRINT) assert _authorized().is_authorized(PEER_FINGERPRINT) is False source = load_journal_source_by_fingerprint(PEER_FINGERPRINT) assert source is not None @@ -200,7 +275,7 @@ def test_unpair_peer_already_revoked_removes_authorized_and_warns( response = _post_unpair(env, {"fingerprint": PEER_FINGERPRINT}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": PEER_FINGERPRINT} + assert response.get_json() == _unpair_payload(PEER_FINGERPRINT) assert _authorized().is_authorized(PEER_FINGERPRINT) is False source = load_journal_source_by_fingerprint(PEER_FINGERPRINT) assert source is not None @@ -221,7 +296,7 @@ def test_unpair_peer_missing_source_removes_authorized_and_warns( response = _post_unpair(env, {"fingerprint": PEER_FINGERPRINT}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": PEER_FINGERPRINT} + assert response.get_json() == _unpair_payload(PEER_FINGERPRINT) assert _authorized().is_authorized(PEER_FINGERPRINT) is False assert load_journal_source_by_fingerprint(PEER_FINGERPRINT) is None assert "peer journal source missing" in caplog.text @@ -245,7 +320,7 @@ def test_unpair_peer_save_failure_removes_authorized_and_logs_error( response = _post_unpair(env, {"device_label": "peer-save-fails"}) assert response.status_code == 200 - assert response.get_json() == {"unpaired": PEER_FINGERPRINT} + assert response.get_json() == _unpair_payload(PEER_FINGERPRINT) assert _authorized().is_authorized(PEER_FINGERPRINT) is False source = load_journal_source_by_fingerprint(PEER_FINGERPRINT) assert source is not None diff --git a/solstone/apps/observer/utils.py b/solstone/apps/observer/utils.py index b330cf8e5..ea98a49dd 100644 --- a/solstone/apps/observer/utils.py +++ b/solstone/apps/observer/utils.py @@ -92,6 +92,11 @@ def observer_device_binding(record: dict[str, Any]) -> dict[str, str] | None: return None +def observer_device_binding_kind(record: dict[str, Any]) -> str | None: + binding = observer_device_binding(record) + return None if binding is None else binding["kind"] + + def get_observers_dir(*, ensure_exists: bool = True) -> Path: """Get the observers storage directory.""" return get_app_storage_path("observer", "observers", ensure_exists=ensure_exists) @@ -515,6 +520,12 @@ def revoke_observer_record(identifier: str) -> dict: return observer +class ObserverRevokeError(RuntimeError): + def __init__(self, message: str, revoked: list[dict]) -> None: + super().__init__(message) + self.revoked = revoked + + def revoke_observers_bound_to_device(device: str) -> list[dict]: """Revoke all observer records bound to a paired-device identity.""" revoked: list[dict] = [] @@ -530,7 +541,7 @@ def revoke_observers_bound_to_device(device: str) -> list[dict]: observer["revoked"] = True observer["revoked_at"] = revoked_at if not save_observer(observer): - raise RuntimeError("failed to save observer") + raise ObserverRevokeError("failed to save observer", revoked) key_prefix = observer_filename_prefix(observer) log_app_action( app="observer", diff --git a/solstone/observe/observer_cli.py b/solstone/observe/observer_cli.py index 7ab2df4d6..46ad3039f 100644 --- a/solstone/observe/observer_cli.py +++ b/solstone/observe/observer_cli.py @@ -30,6 +30,7 @@ from solstone.apps.observer.utils import ( get_hist_dir, list_observers, load_history, + observer_device_binding_kind, observer_filename_prefix, pruned_segments, revoke_observer_record, @@ -189,6 +190,7 @@ def cmd_list(args: argparse.Namespace) -> int: "name": r.get("name", ""), "prefix": observer_filename_prefix(r), "status": _status_label(r), + "device_binding_kind": observer_device_binding_kind(r), "last_seen": r.get("last_seen"), "last_segment_received_at": r.get("last_segment_received_at"), "last_segment_day": r.get("last_segment_day"), @@ -204,22 +206,23 @@ def cmd_list(args: argparse.Namespace) -> int: return 0 print( - f"{'Name':<20} {'Prefix':<18} {'Status':<14} " + f"{'Name':<20} {'Prefix':<18} {'Status':<14} {'Binding':<10} " f"{'Last Seen':<18} {'Last Segment':<12} {'Segments':>10} {'Bytes':>12}" ) - print("-" * 107) + print("-" * 118) for r in observers: name = r.get("name", "") prefix = observer_filename_prefix(r) status = _status_label(r) + binding = observer_device_binding_kind(r) or "unbound" last_seen = _fmt_time(r.get("last_seen")) last_segment = _fmt_compact_age(r.get("last_segment_received_at")) stats = r.get("stats", {}) segments = stats.get("segments_received", 0) bytes_recv = _fmt_bytes(stats.get("bytes_received", 0)) print( - f"{name:<20} {prefix:<18} {status:<14} " + f"{name:<20} {prefix:<18} {status:<14} {binding:<10} " f"{last_seen:<18} {last_segment:<12} {segments:>10} {bytes_recv:>12}" ) @@ -380,6 +383,7 @@ def _status_single(identifier: str, json_output: bool = False) -> int: "name": name, "prefix": key_prefix, "status": _status_label(observer), + "device_binding_kind": observer_device_binding_kind(observer), "created_at": observer.get("created_at"), "last_seen": observer.get("last_seen"), "last_segment_received_at": observer.get( @@ -424,6 +428,7 @@ def _status_single(identifier: str, json_output: bool = False) -> int: print(f"Observer: {name}") print_field("Prefix:", key_prefix) print_field("Status:", _status_label(observer)) + print_field("Binding:", observer_device_binding_kind(observer) or "unbound") print_field("Created:", _fmt_time(observer.get("created_at"))) print_field("Last seen:", _fmt_time(observer.get("last_seen"))) print_field("Last segment:", last_segment_context) @@ -496,6 +501,7 @@ def _status_all(json_output: bool = False) -> int: "name": r.get("name", ""), "prefix": observer_filename_prefix(r), "status": _status_label(r), + "device_binding_kind": observer_device_binding_kind(r), "last_seen": r.get("last_seen"), "last_segment_received_at": r.get( "last_segment_received_at" @@ -517,18 +523,20 @@ def _status_all(json_output: bool = False) -> int: print(f" Total bytes: {_fmt_bytes(total_bytes)}") print( - f"\n{'Name':<20} {'Prefix':<18} {'Status':<14} " + f"\n{'Name':<20} {'Prefix':<18} {'Status':<14} {'Binding':<10} " f"{'Last Seen':<18} {'Last Segment':<12}" ) - print("-" * 87) + print("-" * 98) for r in observers: name = r.get("name", "") prefix = observer_filename_prefix(r) status = _status_label(r) + binding = observer_device_binding_kind(r) or "unbound" last_seen = _fmt_time(r.get("last_seen")) last_segment = _fmt_compact_age(r.get("last_segment_received_at")) print( - f"{name:<20} {prefix:<18} {status:<14} {last_seen:<18} {last_segment:<12}" + f"{name:<20} {prefix:<18} {status:<14} {binding:<10} " + f"{last_seen:<18} {last_segment:<12}" ) return 0 diff --git a/solstone/think/capture_health.py b/solstone/think/capture_health.py index 01d178693..b7ff83eb1 100644 --- a/solstone/think/capture_health.py +++ b/solstone/think/capture_health.py @@ -35,6 +35,7 @@ def get_capture_health() -> dict: get_active_ingest_rejection, get_health_beacon, list_observers, + observer_device_binding_kind, ) try: @@ -73,6 +74,7 @@ def get_capture_health() -> dict: "name": o.get("name", "unknown"), "last_seen": last_seen, "status": obs_status, + "device_binding_kind": observer_device_binding_kind(o), } rejection = get_active_ingest_rejection(o) diff --git a/solstone/think/doctor.py b/solstone/think/doctor.py index 8662034f1..c0e39b525 100644 --- a/solstone/think/doctor.py +++ b/solstone/think/doctor.py @@ -157,6 +157,7 @@ JOURNAL_CAUGHT_UP_CHECK = Check("journal_caught_up", "advisory", ("linux", "darw JOURNAL_MAINT_TASKS_CHECK = Check("journal_maint_tasks", "blocker", ("linux", "darwin")) TASK_PACE_CHECK = Check("task_pace", "advisory", ("linux", "darwin")) CAPTURE_HEALTH_CHECK = Check("capture_health", "advisory", ("linux", "darwin")) +OBSERVER_BINDING_CHECK = Check("observer_binding", "advisory", ("linux", "darwin")) OBSERVER_INGEST_HEALTH_CHECK = Check( "observer_ingest_health", "advisory", ("linux", "darwin") ) @@ -1115,6 +1116,30 @@ def capture_health_check(args: Args) -> CheckResult: ) +def observer_binding_check(args: Args) -> CheckResult: + del args + result = get_capture_health() + observers = result.get("observers", []) + total = len(observers) + unbound = [ + observer + for observer in observers + if observer.get("device_binding_kind") is None + ] + if not unbound: + return make_result( + OBSERVER_BINDING_CHECK, + "ok", + f"active observer records={total}; unbound=0", + ) + names = ", ".join(str(observer.get("name", "unknown")) for observer in unbound) + return make_result( + OBSERVER_BINDING_CHECK, + "ok", + f"active observer records={total}; unbound={len(unbound)}; streams={names}", + ) + + def _observer_delivery_stall_clause( observer: dict, facts: dict, @@ -1475,6 +1500,7 @@ JOURNAL_CHECKS: list[tuple[Check, Runner]] = [ (TASK_PACE_CHECK, task_pace_check), (BRAIN_CHECK, brain_check), (CAPTURE_HEALTH_CHECK, capture_health_check), + (OBSERVER_BINDING_CHECK, observer_binding_check), (OBSERVER_DELIVERY_STALL_CHECK, observer_delivery_stall_check), (OBSERVER_INGEST_HEALTH_CHECK, observer_ingest_health_check), (ORPHAN_SEGMENT_PDF_CHECK, orphan_segment_pdf_check), diff --git a/tests/test_capture_health.py b/tests/test_capture_health.py index 791d101b1..28f43fb7d 100644 --- a/tests/test_capture_health.py +++ b/tests/test_capture_health.py @@ -33,6 +33,7 @@ def test_no_last_seen_is_offline(monkeypatch): result = get_capture_health() assert result["observers"][0]["status"] == "offline" + assert result["observers"][0]["device_binding_kind"] == "cert" def test_disabled_observers_excluded(monkeypatch): @@ -89,6 +90,7 @@ def test_degraded_status_from_rejection(monkeypatch): assert result["status"] == "degraded" observer = result["observers"][0] assert observer["status"] == "degraded" + assert observer["device_binding_kind"] == "cert" assert observer["ingest_rejection"]["reason_code"] == "ingest_contract_invalid" assert "segment" not in observer["ingest_rejection"] @@ -104,6 +106,7 @@ def test_legacy_observer_not_failed(monkeypatch): assert result["status"] == "active" assert result["observers"][0]["status"] == "active" + assert result["observers"][0]["device_binding_kind"] == "cert" assert "ingest_rejection" not in result["observers"][0] monkeypatch.setattr( @@ -122,6 +125,7 @@ def test_legacy_observer_not_failed(monkeypatch): assert result["status"] == "active" assert result["observers"][0]["status"] == "active" + assert result["observers"][0]["device_binding_kind"] == "cert" assert "ingest_rejection" not in result["observers"][0] assert result["observers"][0]["beacon"]["version"] == "0.3.1" @@ -138,4 +142,23 @@ def test_unbound_observer_uses_freshness_status(monkeypatch): assert result["status"] == "active" observer = result["observers"][0] assert observer["status"] == "active" - assert "unbound" not in observer + assert observer["device_binding_kind"] is None + + +def test_all_unbound_active_rolls_up_to_active(monkeypatch): + monkeypatch.setattr("solstone.think.capture_health.now_ms", lambda: 1000) + monkeypatch.setattr( + "solstone.apps.observer.utils.list_observers", + lambda: [ + {"name": "a", "enabled": True, "last_seen": 1000}, + {"name": "b", "enabled": True, "last_seen": 990}, + ], + ) + + result = get_capture_health() + + assert result["status"] == "active" + assert [observer["device_binding_kind"] for observer in result["observers"]] == [ + None, + None, + ] diff --git a/tests/test_journal_doctor.py b/tests/test_journal_doctor.py index 85797503d..d489b0646 100644 --- a/tests/test_journal_doctor.py +++ b/tests/test_journal_doctor.py @@ -339,6 +339,96 @@ def test_capture_health_check_maps_every_rollup_status(doctor, monkeypatch): assert result.detail == "rollup=unknown; observer records unavailable" +def test_observer_binding_check_counts_active_unbound_records(doctor, monkeypatch): + now = DOCTOR_NOW_MS + monkeypatch.setattr("solstone.think.capture_health.now_ms", lambda: now) + monkeypatch.setattr( + "solstone.apps.observer.utils.list_observers", + lambda: [ + {"name": "unbound-a", "enabled": True, "last_seen": now}, + {"name": "unbound-b", "enabled": True, "last_seen": now}, + bound_observer(name="bound", enabled=True, last_seen=now), + ], + ) + + result = doctor.observer_binding_check(args(doctor)) + + assert result.name == "observer_binding" + assert result.status == "ok" + assert result.detail == ( + "active observer records=3; unbound=2; streams=unbound-a, unbound-b" + ) + + +def test_observer_binding_check_ignores_revoked_unbound_records( + doctor, + monkeypatch, +): + now = DOCTOR_NOW_MS + monkeypatch.setattr("solstone.think.capture_health.now_ms", lambda: now) + monkeypatch.setattr( + "solstone.apps.observer.utils.list_observers", + lambda: [ + {"name": "unbound-a", "enabled": True, "last_seen": now}, + {"name": "revoked-unbound", "revoked": True, "last_seen": now}, + bound_observer(name="bound", enabled=True, last_seen=now), + ], + ) + + result = doctor.observer_binding_check(args(doctor)) + + assert result.status == "ok" + assert result.detail == "active observer records=2; unbound=1; streams=unbound-a" + + +def test_observer_binding_check_zero_observer_and_zero_unbound_wording( + doctor, + monkeypatch, +): + monkeypatch.setattr("solstone.apps.observer.utils.list_observers", lambda: []) + + result = doctor.observer_binding_check(args(doctor)) + + assert result.status == "ok" + assert result.detail == "active observer records=0; unbound=0" + + now = DOCTOR_NOW_MS + monkeypatch.setattr("solstone.think.capture_health.now_ms", lambda: now) + monkeypatch.setattr( + "solstone.apps.observer.utils.list_observers", + lambda: [bound_observer(name="bound", enabled=True, last_seen=now)], + ) + + result = doctor.observer_binding_check(args(doctor)) + + assert result.status == "ok" + assert result.detail == "active observer records=1; unbound=0" + + +def test_observer_binding_check_warning_counts_match_bound_and_unbound( + doctor, + monkeypatch, +): + now = DOCTOR_NOW_MS + monkeypatch.setattr("solstone.think.capture_health.now_ms", lambda: now) + monkeypatch.setattr( + "solstone.apps.observer.utils.list_observers", + lambda: [{"name": "unbound", "enabled": True, "last_seen": now}], + ) + unbound_counts = doctor.summary_counts( + [doctor.observer_binding_check(args(doctor))] + ) + + monkeypatch.setattr( + "solstone.apps.observer.utils.list_observers", + lambda: [bound_observer(name="bound", enabled=True, last_seen=now)], + ) + bound_counts = doctor.summary_counts([doctor.observer_binding_check(args(doctor))]) + + assert unbound_counts["failed"] == bound_counts["failed"] == 0 + assert unbound_counts["warnings"] == bound_counts["warnings"] == 0 + + def test_lockstep_stale_stamps_warn_on_capture_health_only(doctor, monkeypatch): now = DOCTOR_NOW_MS stale_stamp = now - (7 * HOUR_MS) @@ -694,8 +784,14 @@ def test_new_doctor_checks_are_registered(doctor): doctor.OBSERVER_DELIVERY_STALL_CHECK, doctor.observer_delivery_stall_check, ) in doctor.JOURNAL_CHECKS + assert ( + doctor.OBSERVER_BINDING_CHECK, + doctor.observer_binding_check, + ) in doctor.JOURNAL_CHECKS assert doctor.CAPTURE_HEALTH_CHECK.severity == "advisory" assert doctor.CAPTURE_HEALTH_CHECK.platforms == ("linux", "darwin") + assert doctor.OBSERVER_BINDING_CHECK.severity == "advisory" + assert doctor.OBSERVER_BINDING_CHECK.platforms == ("linux", "darwin") assert doctor.OBSERVER_DELIVERY_STALL_CHECK.severity == "advisory" assert doctor.OBSERVER_DELIVERY_STALL_CHECK.platforms == ("linux", "darwin") diff --git a/tests/test_observer_cli.py b/tests/test_observer_cli.py index cae793175..f5f51098c 100644 --- a/tests/test_observer_cli.py +++ b/tests/test_observer_cli.py @@ -440,6 +440,7 @@ def _table_cell(output: str, row: str, column: str) -> str: "Name", "Prefix", "Status", + "Binding", "Last Seen", "Last Segment", "Segments", @@ -756,6 +757,7 @@ def test_cmd_list_json_includes_prefix_and_status( rows = {row["name"]: row for row in json.loads(captured.out)} assert rows["desktop"]["prefix"] == "abcdefgh" assert rows["desktop"]["status"] == "disconnected" + assert rows["desktop"]["device_binding_kind"] == "cert" assert "mode" not in rows["desktop"] @@ -789,7 +791,9 @@ def test_missing_binding_uses_connection_status_in_list_and_status( assert rc == 0 rows = {row["name"]: row for row in json.loads(captured.out)} assert rows["unbound"]["status"] == "connected" + assert rows["unbound"]["device_binding_kind"] is None assert rows["revoked"]["status"] == "revoked" + assert rows["revoked"]["device_binding_kind"] is None rc = observer_cli.cmd_list(argparse.Namespace(json_output=False)) @@ -798,6 +802,9 @@ def test_missing_binding_uses_connection_status_in_list_and_status( assert _table_cell(captured.out, _table_row(captured.out, "unbound"), "Status") == ( "connected" ) + assert _table_cell( + captured.out, _table_row(captured.out, "unbound"), "Binding" + ) == ("unbound") assert _table_cell(captured.out, _table_row(captured.out, "revoked"), "Status") == ( "revoked" ) @@ -808,7 +815,9 @@ def test_missing_binding_uses_connection_status_in_list_and_status( captured = capsys.readouterr() assert rc == 0 - assert json.loads(captured.out)["status"] == "connected" + single = json.loads(captured.out) + assert single["status"] == "connected" + assert single["device_binding_kind"] is None rc = observer_cli.cmd_status( argparse.Namespace(identifier="unbound", json_output=False) @@ -817,6 +826,7 @@ def test_missing_binding_uses_connection_status_in_list_and_status( captured = capsys.readouterr() assert rc == 0 assert " Status: connected\n" in captured.out + assert " Binding: unbound\n" in captured.out rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=True)) @@ -830,6 +840,12 @@ def test_missing_binding_uses_connection_status_in_list_and_status( "unbound": "connected", "revoked": "revoked", } + assert { + row["name"]: row["device_binding_kind"] for row in payload["observers"] + } == { + "unbound": None, + "revoked": None, + } rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=False)) @@ -839,6 +855,9 @@ def test_missing_binding_uses_connection_status_in_list_and_status( assert _table_cell(captured.out, _table_row(captured.out, "unbound"), "Status") == ( "connected" ) + assert _table_cell( + captured.out, _table_row(captured.out, "unbound"), "Binding" + ) == ("unbound") def test_fmt_compact_age_units_and_guards(