Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
from __future__ import annotations
import argparseimport inspectimport jsonimport reimport sqlite3import sysfrom pathlib import Pathfrom types import SimpleNamespace
import pytest
from solstone.apps.observer.prune import format_result, result_exit_code, run_prunefrom solstone.apps.observer.utils import ( append_history_record, list_observers, load_history, revoke_observer_record, save_observer,)from solstone.observe import observer_clifrom solstone.observe.processing_record import ( HANDLER_TRANSCRIBE, SCHEMA, STATE_EMPTY,)from solstone.think.indexer.journal import get_journal_indexfrom solstone.think.link.auth import AuthorizedClientsfrom solstone.think.link.paths import authorized_clients_pathfrom solstone.think.streams import ( get_stream_state, read_segment_stream, write_segment_stream,)from solstone.think.utils import iter_segments
_DEFAULT_DEVICE_BINDING = object()
CREATE_RETIRED_MESSAGE = ( "journal observer create is retired. observers register themselves: from " "this machine directly, or from a paired device.\n" "for a remote device, pair it first: sol call link pair --device-label <name>\n" "if a device was re-paired and its stream is stuck, clear the old observer first:\n" " journal observer revoke <name>\n")
@pytest.fixturedef observer_cli_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): home = tmp_path / "home" journal = tmp_path / "journal" home.mkdir() journal.mkdir() monkeypatch.setattr(Path, "home", lambda: home) monkeypatch.setenv("HOME", str(home)) monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
import solstone.convey.state as convey_state
convey_state.journal_root = "" return SimpleNamespace(home=home, journal=journal)
def _observer_record_paths(journal: Path) -> list[Path]: return sorted((journal / "apps" / "observer" / "observers").glob("*.json"))
def _observer_create_actions(journal: Path) -> list[dict]: actions_dir = journal / "config" / "actions" if not actions_dir.exists(): return [] actions: list[dict] = [] for path in sorted(actions_dir.glob("*.jsonl")): for line in path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue record = json.loads(line) if record.get("action") == "observer_create": actions.append(record) return actions
def _run_observer_main( argv: list[str], monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> tuple[int, pytest.CaptureResult[str]]: monkeypatch.setattr(sys, "argv", argv) with pytest.raises(SystemExit) as exc: observer_cli.main() return int(exc.value.code), capsys.readouterr()
def _observer( name: str = "archon", key: str = "existing-key-abcdef", *, last_seen: int | None = None, last_segment: str | None = None, last_segment_received_at: object = None, last_segment_day: object = None, include_last_segment_freshness: bool = True, device_binding: object = _DEFAULT_DEVICE_BINDING,) -> dict: record = { "key": key, "name": name, "created_at": 1, "last_seen": last_seen, "last_segment": last_segment, "enabled": True, "stats": {"segments_received": 0, "bytes_received": 0}, } if include_last_segment_freshness: record["last_segment_received_at"] = last_segment_received_at record["last_segment_day"] = last_segment_day if device_binding is _DEFAULT_DEVICE_BINDING: record["device_binding"] = { "device": "sha256:" + ("a" * 64), "kind": "cert", } elif device_binding is not None: record["device_binding"] = device_binding return record
PRUNE_DAY = "20250103"PRUNE_STREAM = "field"PRUNE_AUDIO = b"identical legacy audio bytes"PRUNE_SCREEN = b"identical legacy screen bytes"
def _observer_for_stream( stream: str = PRUNE_STREAM, key: str = "field-key-abcdef") -> dict: record = _observer(name=stream, key=key) record["stream"] = stream return record
def _processing_row(size: int) -> str: return ( json.dumps( { "raw": "audio.flac", "_solstone_processing": { "schema": SCHEMA, "state": STATE_EMPTY, "handler": HANDLER_TRANSCRIBE, "input_size": size, }, } ) + "\n" )
def _write_prune_segment( journal: Path, *, segment: str, seq: int, prev_segment: str | None, audio: bytes = PRUNE_AUDIO, screen: bytes = PRUNE_SCREEN, manifest: bool = False, marker: bool = True, unknown_file: bool = False, extra_manifest_content: bool = False, proof_only_audio: bool = False,) -> Path: seg_dir = journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / segment seg_dir.mkdir(parents=True) if marker: write_segment_stream( seg_dir, PRUNE_STREAM, PRUNE_DAY if prev_segment else None, prev_segment, seq, ) if not proof_only_audio: (seg_dir / "audio.flac").write_bytes(audio) (seg_dir / "audio.jsonl").write_text( json.dumps({"segment": segment, "text": f"transcript {segment}"}) + "\n", encoding="utf-8", ) else: (seg_dir / "audio.jsonl").write_text( _processing_row(len(audio)), encoding="utf-8", ) (seg_dir / "screen.mp4").write_bytes(screen) (seg_dir / "screen.jsonl").write_text( json.dumps({"segment": segment, "text": f"description {segment}"}) + "\n", encoding="utf-8", ) (seg_dir / "events.jsonl").write_text( json.dumps({"event": segment}) + "\n", encoding="utf-8", ) talents = seg_dir / "talents" talents.mkdir() (talents / "sense.json").write_text( json.dumps({"segment": segment}), encoding="utf-8", ) if unknown_file: (seg_dir / "notes.txt").write_text("operator note", encoding="utf-8") manifest_files = None if manifest: manifest_files = { "audio.flac": {"sha256": _sha(audio), "size": len(audio)}, "screen.mp4": {"sha256": _sha(screen), "size": len(screen)}, } if extra_manifest_content: extra = b"unique manifest content" (seg_dir / "capture.bin").write_bytes(extra) manifest_files["capture.bin"] = {"sha256": _sha(extra), "size": len(extra)} (seg_dir / "ingest.json").write_text( json.dumps( { "schema_version": 1, "requested_segment": segment, "files": manifest_files, } ) + "\n", encoding="utf-8", ) return seg_dir
def _sha(data: bytes) -> str: import hashlib
return hashlib.sha256(data).hexdigest()
def _append_upload(prefix: str, segment: str, *, sha: str | None = None) -> None: append_history_record( prefix, PRUNE_DAY, { "ts": 1, "segment": segment, "stream": PRUNE_STREAM, "files": [ { "submitted": "audio.flac", "written": "audio.flac", "size": len(PRUNE_AUDIO), "sha256": sha or _sha(PRUNE_AUDIO), } ], }, )
def _write_stream_state(journal: Path, *, last_segment: str, seq: int = 9) -> None: streams = journal / "streams" streams.mkdir() (streams / f"{PRUNE_STREAM}.json").write_text( json.dumps( { "name": PRUNE_STREAM, "type": "observer", "host": "field-host", "platform": "linux", "created_at": 123, "last_day": PRUNE_DAY, "last_segment": last_segment, "seq": seq, } ) + "\n", encoding="utf-8", )
def _insert_index_rows(journal: Path, *segments: str) -> None: conn, _db_path = get_journal_index(str(journal)) try: for segment in segments: rel = f"{PRUNE_DAY}/{PRUNE_STREAM}/{segment}" conn.execute( "INSERT INTO files(path, mtime) VALUES (?, ?)", (f"{rel}/audio.flac", 1) ) conn.execute( "INSERT INTO chunks(content, path, day, facet, agent, stream, idx, time_bucket) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ("content", rel, PRUNE_DAY, "", "segment", PRUNE_STREAM, 0, "morning"), ) conn.commit() finally: conn.close()
def _index_count(journal: Path, segment: str) -> int: db_path = journal / "indexer" / "journal.sqlite" conn = sqlite3.connect(db_path) try: rel = f"{PRUNE_DAY}/{PRUNE_STREAM}/{segment}" return conn.execute( "SELECT count(*) FROM chunks WHERE path = ? OR path LIKE ?", (rel, f"{rel}/%"), ).fetchone()[0] finally: conn.close()
def _journal_snapshot(journal: Path) -> list[tuple[str, str, bytes]]: entries = [] for path in sorted(journal.rglob("*")): rel = path.relative_to(journal).as_posix() if path.is_dir(): entries.append((rel, "dir", b"")) elif path.is_file(): entries.append((rel, "file", path.read_bytes())) return entries
def _stream_marker_snapshot(journal: Path) -> dict[str, dict]: snapshot = {} for stream, segment, path in iter_segments(PRUNE_DAY): if stream != PRUNE_STREAM: continue snapshot[segment] = read_segment_stream(path) return snapshot
def _pruned_history(prefix: str, segment: str) -> list[dict]: return [ record for record in load_history(prefix, PRUNE_DAY) if record.get("type") == "pruned" and record.get("segment") == segment ]
def _build_legacy_prune_lattice(journal: Path, prefix: str) -> None: _write_prune_segment( journal, segment="080000_300", seq=1, prev_segment=None, audio=b"before" ) _write_prune_segment( journal, segment="090000_300", seq=2, prev_segment="080000_300" ) _write_prune_segment( journal, segment="090000_301", seq=3, prev_segment="090000_300" ) _write_prune_segment( journal, segment="090000_302", seq=4, prev_segment="090000_301" ) _write_prune_segment( journal, segment="090000_303", seq=5, prev_segment="090000_302", audio=b"near duplicate audio", ) _write_prune_segment( journal, segment="090000_304", seq=6, prev_segment="090000_303", unknown_file=True, ) _write_prune_segment( journal, segment="100000_300", seq=7, prev_segment="090000_304" ) _write_prune_segment( journal, segment="100500_300", seq=8, prev_segment="100000_300", audio=PRUNE_AUDIO, screen=PRUNE_SCREEN, ) for segment in ( "090000_300", "090000_301", "090000_302", "090000_303", "090000_304", "110000_300", ): _append_upload(prefix, segment) _write_stream_state(journal, last_segment="100500_300") _insert_index_rows(journal, "090000_301", "090000_302")
def _observer_with_stats( *, name: str, key: str, created_at: int, segments_received: int, bytes_received: int, duplicates_rejected: int = 0, last_segment_received_at: object = None, last_segment_day: object = None, include_last_segment_freshness: bool = True,) -> dict: record = _observer( name=name, key=key, last_segment_received_at=last_segment_received_at, last_segment_day=last_segment_day, include_last_segment_freshness=include_last_segment_freshness, ) record["created_at"] = created_at record["stats"] = { "segments_received": segments_received, "bytes_received": bytes_received, "duplicates_rejected": duplicates_rejected, } return record
def _table_row(output: str, name: str) -> str: return next(line for line in output.splitlines() if line.startswith(f"{name:<20}"))
def _table_header(output: str) -> str: return next( line for line in output.splitlines() if line.startswith("Name") and "Last Segment" in line )
def _table_cell(output: str, row: str, column: str) -> str: header = _table_header(output) starts = sorted( header.index(name) for name in ( "Name", "Prefix", "Status", "Binding", "Last Seen", "Last Segment", "Segments", "Bytes", ) if name in header ) start = header.index(column) end = next((pos for pos in starts if pos > start), len(row)) return row[start:end].strip()
def _last_segment_cell(output: str, row: str) -> str: return _table_cell(output, row, "Last Segment")
def test_create_command_refuses_with_pairing_guidance_and_writes_nothing( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: code, captured = _run_observer_main( ["journal observer", "create", "archon"], monkeypatch, capsys, )
assert code == 2 assert captured.out == "" assert captured.err.encode() == CREATE_RETIRED_MESSAGE.encode() assert _observer_record_paths(observer_cli_env.journal) == [] assert _observer_create_actions(observer_cli_env.journal) == []
def test_create_command_accepts_json_flag_but_keeps_retired_stderr_contract( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: code, captured = _run_observer_main( ["journal observer", "create", "archon", "--json"], monkeypatch, capsys, )
assert code == 2 assert captured.out == "" assert captured.err.encode() == CREATE_RETIRED_MESSAGE.encode() assert _observer_record_paths(observer_cli_env.journal) == [] assert _observer_create_actions(observer_cli_env.journal) == []
def test_bare_create_command_refuses_without_argparse_dump( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: code, captured = _run_observer_main( ["journal observer", "create"], monkeypatch, capsys, )
assert code == 2 assert captured.out == "" assert captured.err.encode() == CREATE_RETIRED_MESSAGE.encode()
def test_create_command_is_listed_in_help( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: code, captured = _run_observer_main( ["journal observer", "--help"], monkeypatch, capsys, )
assert code == 0 assert "create" in captured.out assert "Explain retired manual observer creation" in captured.out assert captured.err == ""
def test_create_command_short_circuits_before_solstone_requirement( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: def fail_require_solstone() -> None: raise AssertionError("require_solstone should not be called for create")
monkeypatch.delenv("SOL_SKIP_SUPERVISOR_CHECK", raising=False) monkeypatch.setattr(observer_cli, "require_solstone", fail_require_solstone)
code, captured = _run_observer_main( ["journal observer", "create", "archon"], monkeypatch, capsys, )
assert code == 2 assert captured.out == "" assert captured.err.encode() == CREATE_RETIRED_MESSAGE.encode()
def test_reconcile_collapses_duplicates_oldest_survives(observer_cli_env) -> None: assert save_observer( _observer_with_stats( name="fedora.tmux", key="newest03-key", created_at=3, segments_received=5, bytes_received=100, duplicates_rejected=1, ) ) assert save_observer( _observer_with_stats( name="fedora.tmux", key="oldest01-key", created_at=1, segments_received=7, bytes_received=200, duplicates_rejected=2, ) ) assert save_observer( _observer_with_stats( name="fedora.tmux", key="middle02-key", created_at=2, segments_received=11, bytes_received=300, ) ) lone = _observer_with_stats( name="fedora", key="desktop1-key", created_at=4, segments_received=13, bytes_received=400, duplicates_rejected=5, ) assert save_observer(lone)
plan = observer_cli.reconcile_observers(dry_run=False)
assert plan == [ { "name": "fedora.tmux", "survivor_prefix": "oldest01", "revoked_prefixes": ["newest03", "middle02"], "stats": { "segments_received": 23, "bytes_received": 600, "duplicates_rejected": 3, }, } ] records = list_observers() tmux_records = [record for record in records if record["name"] == "fedora.tmux"] unrevoked_tmux = [ record for record in tmux_records if not record.get("revoked", False) ] assert len(unrevoked_tmux) == 1 assert unrevoked_tmux[0]["created_at"] == 1 assert unrevoked_tmux[0]["stats"] == { "segments_received": 23, "bytes_received": 600, "duplicates_rejected": 3, } revoked_tmux = [record for record in tmux_records if record.get("revoked", False)] assert {record["created_at"] for record in revoked_tmux} == {2, 3} lone_record = next(record for record in records if record["name"] == "fedora") assert lone_record.get("revoked", False) is False assert lone_record["stats"] == lone["stats"]
def test_reconcile_dry_run_mutates_nothing(observer_cli_env) -> None: assert save_observer( _observer_with_stats( name="fedora.tmux", key="newest03-key", created_at=3, segments_received=5, bytes_received=100, duplicates_rejected=1, ) ) assert save_observer( _observer_with_stats( name="fedora.tmux", key="oldest01-key", created_at=1, segments_received=7, bytes_received=200, duplicates_rejected=2, ) ) assert save_observer( _observer_with_stats( name="fedora.tmux", key="middle02-key", created_at=2, segments_received=11, bytes_received=300, ) ) observers_dir = observer_cli_env.journal / "apps" / "observer" / "observers" before = {path.name: path.read_bytes() for path in observers_dir.glob("*.json")}
plan = observer_cli.reconcile_observers(dry_run=True)
assert plan == [ { "name": "fedora.tmux", "survivor_prefix": "oldest01", "revoked_prefixes": ["newest03", "middle02"], "stats": { "segments_received": 23, "bytes_received": 600, "duplicates_rejected": 3, }, } ] after = {path.name: path.read_bytes() for path in observers_dir.glob("*.json")} assert after == before
def test_reconcile_lone_stream_returns_empty_plan(observer_cli_env) -> None: lone = _observer_with_stats( name="fedora", key="desktop1-key", created_at=1, segments_received=13, bytes_received=400, duplicates_rejected=5, ) assert save_observer(lone)
plan = observer_cli.reconcile_observers(dry_run=False)
assert plan == [] records = list_observers() assert len(records) == 1 assert records[0].get("revoked", False) is False assert records[0]["stats"] == lone["stats"]
def test_cmd_reconcile_reports_plan( observer_cli_env, capsys: pytest.CaptureFixture[str],) -> None: assert save_observer( _observer_with_stats( name="fedora.tmux", key="newest03-key", created_at=3, segments_received=5, bytes_received=100, ) ) assert save_observer( _observer_with_stats( name="fedora.tmux", key="oldest01-key", created_at=1, segments_received=7, bytes_received=200, ) )
rc = observer_cli.cmd_reconcile( argparse.Namespace(dry_run=False, json_output=False) )
captured = capsys.readouterr() assert rc == 0 assert captured.err == "" assert "Reconciled stream 'fedora.tmux':" in captured.out assert " survivor: oldest01" in captured.out assert " revoking: newest03" in captured.out
def test_cmd_reconcile_no_duplicates( observer_cli_env, capsys: pytest.CaptureFixture[str],) -> None: assert save_observer(_observer(name="fedora", key="desktop1-key"))
rc = observer_cli.cmd_reconcile( argparse.Namespace(dry_run=False, json_output=False) )
captured = capsys.readouterr() assert rc == 0 assert captured.err == "" assert captured.out == "No duplicate observer streams to reconcile.\n"
def test_cmd_list_json_includes_prefix_and_status( observer_cli_env, capsys: pytest.CaptureFixture[str],) -> None: assert save_observer(_observer(name="desktop", key="abcdefgh12345678")) args = argparse.Namespace(json_output=True)
rc = observer_cli.cmd_list(args)
captured = capsys.readouterr() assert rc == 0 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"]
def test_missing_binding_uses_connection_status_in_list_and_status( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert save_observer( _observer( name="unbound", key="unbound12345678", last_seen=now - 1_000, device_binding=None, ) ) revoked = _observer( name="revoked", key="revoked12345678", last_seen=now - 1_000, device_binding=None, ) revoked["revoked"] = True assert save_observer(revoked)
rc = observer_cli.cmd_list(argparse.Namespace(json_output=True))
captured = capsys.readouterr() 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))
captured = capsys.readouterr() assert rc == 0 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" )
rc = observer_cli.cmd_status( argparse.Namespace(identifier="unbound", json_output=True) )
captured = capsys.readouterr() assert rc == 0 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) )
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))
captured = capsys.readouterr() assert rc == 0 payload = json.loads(captured.out) assert payload["revoked"] == 1 assert payload["connected"] == 1 assert "unbound" not in payload assert {row["name"]: row["status"] for row in payload["observers"]} == { "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))
captured = capsys.readouterr() assert rc == 0 assert " Unbound:" not in captured.out 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( monkeypatch: pytest.MonkeyPatch,) -> None: monkeypatch.setattr(observer_cli, "now_ms", lambda: 0) assert observer_cli._fmt_compact_age(0) == "0s"
now = 2_000_000_000_000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert observer_cli._fmt_compact_age(None) == "—" assert observer_cli._fmt_compact_age("bad") == "—" assert observer_cli._fmt_compact_age(True) == "—" assert observer_cli._fmt_compact_age(-1) == "—" assert observer_cli._fmt_compact_age(now + 1) == "—" assert observer_cli._fmt_compact_age(now) == "0s" assert observer_cli._fmt_compact_age(now - 30_000) == "30s" assert observer_cli._fmt_compact_age(now - ((59 * 60 + 59) * 1000)) == "59m" assert observer_cli._fmt_compact_age(now - (60 * 60 * 1000)) == "1h" assert observer_cli._fmt_compact_age(now - ((23 * 60 + 59) * 60 * 1000)) == "23h" assert observer_cli._fmt_compact_age(now - (24 * 60 * 60 * 1000)) == "1d" assert observer_cli._fmt_compact_age(now - int(19.5 * 60 * 60 * 1000)) == "19h"
def test_cmd_list_shows_last_segment_column_and_json( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert save_observer( _observer( name="desktop", key="abcdefgh12345678", last_segment_received_at=now - 2 * 60 * 1000, last_segment_day="20260724", ) )
rc = observer_cli.cmd_list(argparse.Namespace(json_output=False))
captured = capsys.readouterr() assert rc == 0 assert "Last Seen Last Segment" in captured.out assert "-" * 107 in captured.out assert _last_segment_cell(captured.out, _table_row(captured.out, "desktop")) == "2m"
rc = observer_cli.cmd_list(argparse.Namespace(json_output=True))
captured = capsys.readouterr() assert rc == 0 rows = {row["name"]: row for row in json.loads(captured.out)} assert rows["desktop"]["last_segment_received_at"] == now - 2 * 60 * 1000 assert rows["desktop"]["last_segment_day"] == "20260724"
def test_cmd_list_human_shows_prefix_column( observer_cli_env, capsys: pytest.CaptureFixture[str],) -> None: assert save_observer(_observer(name="desktop", key="abcdefgh12345678")) args = argparse.Namespace(json_output=False)
rc = observer_cli.cmd_list(args)
captured = capsys.readouterr() assert rc == 0 assert "Name Prefix" in captured.out assert "Mode" not in captured.out assert "desktop abcdefgh" in captured.out
def test_cmd_status_single_reports_prefix( observer_cli_env, capsys: pytest.CaptureFixture[str],) -> None: assert save_observer(_observer(name="desktop", key="cdefghij12345678"))
rc = observer_cli.cmd_status( argparse.Namespace(identifier="desktop", json_output=True) )
captured = capsys.readouterr() assert rc == 0 payload = json.loads(captured.out) assert payload["prefix"] == "cdefghij" assert payload["status"] == "disconnected" assert "mode" not in payload
def test_cmd_status_single_last_segment_age_uses_receipt_time_with_day_context( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 received_at = now - 2 * 60 * 1000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert save_observer( _observer( name="desktop", key="cdefghij12345678", last_seen=now - 1_000, last_segment="120000_300", last_segment_received_at=received_at, last_segment_day="20260722", ) )
rc = observer_cli.cmd_status( argparse.Namespace(identifier="desktop", json_output=False) )
captured = capsys.readouterr() assert rc == 0 assert " Last segment: 2m (20260722)\n" in captured.out
rc = observer_cli.cmd_status( argparse.Namespace(identifier="desktop", json_output=True) )
captured = capsys.readouterr() assert rc == 0 payload = json.loads(captured.out) assert payload["last_segment_received_at"] == received_at assert payload["last_segment_day"] == "20260722"
def test_cmd_status_all_table_shows_prefix( observer_cli_env, capsys: pytest.CaptureFixture[str],) -> None: assert save_observer(_observer(name="desktop", key="abcdefgh12345678"))
rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=False))
captured = capsys.readouterr() assert rc == 0 assert "Name Prefix" in captured.out assert "Mode" not in captured.out assert "desktop abcdefgh" in captured.out
def test_cmd_status_all_shows_last_segment_column_and_json( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert save_observer( _observer( name="desktop", key="abcdefgh12345678", last_segment_received_at=now - 2 * 60 * 1000, last_segment_day="20260724", ) )
rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=False))
captured = capsys.readouterr() assert rc == 0 assert "Last Seen Last Segment" in captured.out assert "-" * 87 in captured.out assert _last_segment_cell(captured.out, _table_row(captured.out, "desktop")) == "2m"
rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=True))
captured = capsys.readouterr() assert rc == 0 payload = json.loads(captured.out) row = payload["observers"][0] assert row["last_segment_received_at"] == now - 2 * 60 * 1000 assert row["last_segment_day"] == "20260724"
def test_last_segment_freshness_does_not_change_connection_status( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert observer_cli.CONNECTED_THRESHOLD_MS == 2 * 60 * 1000 assert save_observer( _observer( name="desktop", key="abcdefgh12345678", last_seen=now - 1_000, last_segment_received_at=now - 41 * 24 * 60 * 60 * 1000, last_segment_day="20260613", ) )
rc = observer_cli.cmd_status( argparse.Namespace(identifier="desktop", json_output=False) )
captured = capsys.readouterr() assert rc == 0 assert " Status: connected\n" in captured.out assert " Last segment: 41d (20260613)\n" in captured.out
def test_fleet_views_show_fresh_last_seen_with_unknown_last_segment( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert save_observer( _observer( name="desktop", key="abcdefgh12345678", last_seen=now - 30_000, include_last_segment_freshness=False, ) )
def assert_divergent_row(output: str) -> None: row = _table_row(output, "desktop") assert _table_cell(output, row, "Status") == "connected" assert _table_cell(output, row, "Last Seen") != "never" assert _last_segment_cell(output, row) == "—"
rc = observer_cli.cmd_list(argparse.Namespace(json_output=False))
captured = capsys.readouterr() assert rc == 0 assert_divergent_row(captured.out)
rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=False))
captured = capsys.readouterr() assert rc == 0 assert_divergent_row(captured.out)
def test_observer_cli_last_segment_rendering_has_no_classification() -> None: source = Path(observer_cli.__file__).read_text(encoding="utf-8") thresholds = re.findall( r"^([A-Z][A-Z0-9_]*THRESHOLD[A-Z0-9_]*)\s*=", source, flags=re.MULTILINE ) assert thresholds == ["CONNECTED_THRESHOLD_MS"]
render_source = "\n".join( inspect.getsource(obj) for obj in ( observer_cli.cmd_list, observer_cli._status_single, observer_cli._status_all, observer_cli._fmt_compact_age, ) ) assert "\\x1b" not in render_source assert "\\033" not in render_source assert "color" not in render_source.lower() assert "colour" not in render_source.lower() assert "stale" not in render_source.lower() for glyph in ("▲", "▼", "●", "○", "✕", "✓"): assert glyph not in render_source
def test_fleet_last_segment_bad_rows_are_isolated( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 monkeypatch.setattr(observer_cli, "now_ms", lambda: now) records = [ _observer( name="good", key="good000012345678", last_segment_received_at=now - 2 * 60 * 1000, ), _observer( name="malformed", key="malform12345678", last_segment_received_at="bad", ), _observer( name="negative", key="negative12345678", last_segment_received_at=-1, ), _observer( name="future", key="future0012345678", last_segment_received_at=now + 1, ), ] for record in records: assert save_observer(record)
rc = observer_cli.cmd_list(argparse.Namespace(json_output=False))
captured = capsys.readouterr() assert rc == 0 assert _last_segment_cell(captured.out, _table_row(captured.out, "good")) == "2m" assert ( _last_segment_cell(captured.out, _table_row(captured.out, "malformed")) == "—" ) assert _last_segment_cell(captured.out, _table_row(captured.out, "negative")) == "—" assert _last_segment_cell(captured.out, _table_row(captured.out, "future")) == "—"
rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=False))
captured = capsys.readouterr() assert rc == 0 assert _last_segment_cell(captured.out, _table_row(captured.out, "good")) == "2m" assert ( _last_segment_cell(captured.out, _table_row(captured.out, "malformed")) == "—" ) assert _last_segment_cell(captured.out, _table_row(captured.out, "negative")) == "—" assert _last_segment_cell(captured.out, _table_row(captured.out, "future")) == "—"
def test_prechange_record_uses_status_single_history_fallback_only( observer_cli_env, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str],) -> None: now = 2_000_000_000_000 today = observer_cli.datetime.date.today().strftime("%Y%m%d") monkeypatch.setattr(observer_cli, "now_ms", lambda: now) assert save_observer( _observer( name="desktop", key="abcdefgh12345678", last_segment="120000_300", include_last_segment_freshness=False, ) ) append_history_record( "abcdefgh", today, { "ts": now - 2 * 60 * 1000, "segment": "120000_300", "stream": "desktop", "files": [], }, ) append_history_record( "abcdefgh", today, {"type": "observed", "ts": now - 10_000, "segment": "ignored"}, )
rc = observer_cli.cmd_list(argparse.Namespace(json_output=False))
captured = capsys.readouterr() assert rc == 0 assert _last_segment_cell(captured.out, _table_row(captured.out, "desktop")) == "—"
rc = observer_cli.cmd_status(argparse.Namespace(identifier=None, json_output=False))
captured = capsys.readouterr() assert rc == 0 assert _last_segment_cell(captured.out, _table_row(captured.out, "desktop")) == "—"
rc = observer_cli.cmd_status( argparse.Namespace(identifier="desktop", json_output=False) )
captured = capsys.readouterr() assert rc == 0 assert f" Last segment: 2m ({today})\n" in captured.out
rc = observer_cli.cmd_status( argparse.Namespace(identifier="desktop", json_output=True) )
captured = capsys.readouterr() assert rc == 0 payload = json.loads(captured.out) assert payload["last_segment_received_at"] is None assert payload["last_segment_day"] is None
def test_revoke_dl_observer_leaves_authorized_clients_untouched( observer_cli_env,) -> None: assert save_observer(_observer(name="desktop", key="abcdefgh12345678")) fingerprint = "sha256:" + ("f" * 64) authorized = AuthorizedClients(authorized_clients_path()) authorized.add( fingerprint, "phone", "inst-1", paired_at="2026-05-20T00:00:00Z", ) before = authorized_clients_path().read_bytes()
record = revoke_observer_record("desktop")
assert record["revoked"] is True assert authorized_clients_path().read_bytes() == before assert ( AuthorizedClients(authorized_clients_path()).is_authorized(fingerprint) is True )
def test_prune_dry_run_lists_legacy_duplicates_and_writes_nothing( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _build_legacy_prune_lattice(observer_cli_env.journal, prefix)
before_history = load_history(prefix, PRUNE_DAY) result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=False)
assert len(result.groups) == 1 assert [c.analysis.segment for c in result.groups[0].candidates] == [ "090000_301", "090000_302", ] assert {refusal.gate for refusal in result.refusals} == { "content-identity", "derived-output", } assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "090000_301" ).is_dir() assert load_history(prefix, PRUNE_DAY) == before_history assert not (observer_cli_env.journal / "chronicle" / PRUNE_DAY / "health").exists() assert _index_count(observer_cli_env.journal, "090000_301") == 1
def test_prune_dry_run_without_observer_storage_writes_nothing( observer_cli_env,) -> None: _write_prune_segment( observer_cli_env.journal, segment="081000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="081000_301", seq=2, prev_segment="081000_300", ) _write_stream_state(observer_cli_env.journal, last_segment="081000_301", seq=2) assert not (observer_cli_env.journal / "apps").exists() before = _journal_snapshot(observer_cli_env.journal)
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=False)
assert any(refusal.gate == "observer-attribution" for refusal in result.refusals) assert result.deleted == [] assert _journal_snapshot(observer_cli_env.journal) == before
def test_prune_execute_deletes_duplicates_repairs_chain_history_and_index( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _build_legacy_prune_lattice(observer_cli_env.journal, prefix) canonical = ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "090000_300" ) canonical_hash = _sha((canonical / "audio.flac").read_bytes())
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert {candidate.analysis.segment for candidate in result.deleted} == { "090000_301", "090000_302", } assert result.crash_repaired == 0 assert result.chain_repaired == 1 output = format_result(result) assert "chain-repaired: 1" in output assert "crash-repaired:" not in output assert result.refusals assert (canonical / "audio.flac").is_file() assert _sha((canonical / "audio.flac").read_bytes()) == canonical_hash assert not ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "090000_301" ).exists() assert not ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "090000_302" ).exists() history = load_history(prefix, PRUNE_DAY) pruned = [row for row in history if row.get("type") == "pruned"] assert {row["segment"] for row in pruned} == {"090000_301", "090000_302"} assert all(row["duplicate_of"] == "090000_300" for row in pruned) assert ( read_segment_stream( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "090000_303" )["prev_segment"] == "090000_300" ) assert _index_count(observer_cli_env.journal, "090000_301") == 0 assert _index_count(observer_cli_env.journal, "090000_302") == 0 assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / "health" / "stream.updated" ).exists()
state = get_stream_state(PRUNE_STREAM) assert state["type"] == "observer" assert state["host"] == "field-host" assert state["platform"] == "linux" assert state["created_at"] == 123 assert state["last_segment"] == "100500_300" assert state["seq"] == 9
existing = { segment for _stream, segment, _path in iter_segments(PRUNE_DAY) if _stream == PRUNE_STREAM } for _stream, _segment, path in iter_segments(PRUNE_DAY): marker = read_segment_stream(path) if marker and marker.get("prev_segment"): assert marker["prev_segment"] in existing
def test_prune_same_start_grouping_does_not_delete_different_start_identical_bytes( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="120000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="120500_300", seq=2, prev_segment="120000_300", ) _append_upload(prefix, "120000_300") _append_upload(prefix, "120500_300") _write_stream_state(observer_cli_env.journal, last_segment="120500_300", seq=2)
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert result.groups == [] assert result.deleted == [] assert result.refusals == [] assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "120500_300" ).is_dir()
def test_prune_near_duplicate_media_bytes_refuse_content_identity( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="125000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="125000_301", seq=2, prev_segment="125000_300", ) _write_prune_segment( observer_cli_env.journal, segment="125000_302", seq=3, prev_segment="125000_301", audio=b"near duplicate audio bytes", ) _append_upload(prefix, "125000_300") _append_upload(prefix, "125000_301") _append_upload(prefix, "125000_302", sha=_sha(b"near duplicate audio bytes")) _write_stream_state(observer_cli_env.journal, last_segment="125000_302", seq=3)
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert [candidate.analysis.segment for candidate in result.deleted] == [ "125000_301" ] refusal = next( refusal for refusal in result.refusals if refusal.subject.endswith("/125000_302") ) assert refusal.gate == "content-identity" assert refusal.file == "audio.flac" assert "compared to canonical 125000_300" in refusal.resolution assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "125000_302" ).is_dir()
def test_prune_markerless_candidate_refuses( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="130000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="130000_301", seq=2, prev_segment="130000_300", marker=False, ) _append_upload(prefix, "130000_300") _append_upload(prefix, "130000_301") _write_stream_state(observer_cli_env.journal, last_segment="130000_301", seq=2)
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert result.deleted == [] assert any(refusal.gate == "chain-identity" for refusal in result.refusals) assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "130000_301" ).is_dir()
def test_prune_unrecognized_derived_file_refuses_legacy_and_manifest_candidates( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="132000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="132000_301", seq=2, prev_segment="132000_300", unknown_file=True, ) _write_prune_segment( observer_cli_env.journal, segment="132500_300", seq=3, prev_segment="132000_301", manifest=True, ) _write_prune_segment( observer_cli_env.journal, segment="132500_301", seq=4, prev_segment="132500_300", manifest=True, unknown_file=True, ) for segment in ("132000_300", "132000_301", "132500_300", "132500_301"): _append_upload(prefix, segment) _write_stream_state(observer_cli_env.journal, last_segment="132500_301", seq=4)
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
refusals = { refusal.subject.rsplit("/", 1)[1]: refusal for refusal in result.refusals if refusal.gate == "derived-output" } assert set(refusals) == {"132000_301", "132500_301"} for refusal in refusals.values(): assert refusal.file == "notes.txt" assert "remove the file" in refusal.resolution assert result.deleted == [] assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "132000_301" ).is_dir() assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "132500_301" ).is_dir()
def test_prune_refuses_manifest_content_name_outside_segment( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="133000_300", seq=1, prev_segment=None, manifest=True, ) candidate = _write_prune_segment( observer_cli_env.journal, segment="133000_301", seq=2, prev_segment="133000_300", manifest=True, ) (candidate.parent / "outside.flac").write_bytes(PRUNE_AUDIO) manifest_path = candidate / "ingest.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["files"]["../outside.flac"] = { "sha256": _sha(PRUNE_AUDIO), "size": len(PRUNE_AUDIO), } manifest_path.write_text(json.dumps(manifest) + "\n", encoding="utf-8") _append_upload(prefix, "133000_300") _append_upload(prefix, "133000_301") _write_stream_state(observer_cli_env.journal, last_segment="133000_301", seq=2)
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
refusal = next( refusal for refusal in result.refusals if refusal.subject.endswith("/133000_301") ) assert refusal.gate == "canonical-heldness" assert refusal.file == "../outside.flac" assert "plain in-segment names" in refusal.resolution assert candidate.is_dir()
def test_prune_manifest_extra_content_refuses_as_content_identity( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="135000_300", seq=1, prev_segment=None, manifest=True, ) _write_prune_segment( observer_cli_env.journal, segment="135000_301", seq=2, prev_segment="135000_300", manifest=True, ) _write_prune_segment( observer_cli_env.journal, segment="135000_302", seq=3, prev_segment="135000_301", manifest=True, extra_manifest_content=True, ) _append_upload(prefix, "135000_300") _append_upload(prefix, "135000_301") _append_upload(prefix, "135000_302") _write_stream_state(observer_cli_env.journal, last_segment="135000_302", seq=3)
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert [candidate.analysis.segment for candidate in result.deleted] == [ "135000_301" ] refusal = next( refusal for refusal in result.refusals if refusal.subject.endswith("/135000_302") ) assert refusal.gate == "content-identity" assert refusal.file == "capture.bin" assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "135000_302" ).is_dir()
def test_prune_refuses_when_canonical_heldness_becomes_unverifiable( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="140000_300", seq=1, prev_segment=None, manifest=True, ) _write_prune_segment( observer_cli_env.journal, segment="140000_301", seq=2, prev_segment="140000_300", manifest=True, ) _append_upload(prefix, "140000_300") _append_upload(prefix, "140000_301") _write_stream_state(observer_cli_env.journal, last_segment="140000_301", seq=2) dry = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=False) assert len(dry.groups) == 1 assert dry.refusals == [] assert [candidate.analysis.segment for candidate in dry.groups[0].candidates] == [ "140000_301" ] canonical_audio = ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "140000_300" / "audio.flac" ) canonical_audio.unlink()
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert result.deleted == [] assert result_exit_code(result) == 2 assert any(refusal.gate == "canonical-heldness" for refusal in result.refusals) assert ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "140000_301" ).is_dir()
def test_prune_last_physical_copy_is_labeled_in_dry_run_and_execute( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="150000_300", seq=1, prev_segment=None, manifest=True, proof_only_audio=True, ) _write_prune_segment( observer_cli_env.journal, segment="150000_301", seq=2, prev_segment="150000_300", manifest=True, ) _append_upload(prefix, "150000_300") _append_upload(prefix, "150000_301") _write_stream_state(observer_cli_env.journal, last_segment="150000_301", seq=2)
dry = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=False) assert dry.last_physical_copy_count == 1 assert dry.groups[0].candidates[0].last_physical_copy is True
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert result.last_physical_copy_count == 1 assert result.deleted[0].last_physical_copy is True
def test_prune_crash_after_pruned_record_before_delete_converges( observer_cli_env, monkeypatch: pytest.MonkeyPatch,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="155000_300", seq=1, prev_segment=None, ) candidate = _write_prune_segment( observer_cli_env.journal, segment="155000_301", seq=2, prev_segment="155000_300", ) _append_upload(prefix, "155000_300") _append_upload(prefix, "155000_301") _write_stream_state(observer_cli_env.journal, last_segment="155000_301", seq=2)
from solstone.apps.observer import prune
def crash_before_delete(_path: Path) -> None: raise RuntimeError("interrupt before delete")
with monkeypatch.context() as patch: patch.setattr(prune.shutil, "rmtree", crash_before_delete) with pytest.raises(RuntimeError, match="interrupt before delete"): run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert candidate.is_dir() assert (candidate / "audio.flac").read_bytes() == PRUNE_AUDIO assert len(_pruned_history(prefix, "155000_301")) == 1
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert result.refusals == [] assert result_exit_code(result) == 0 assert not candidate.exists() assert len(_pruned_history(prefix, "155000_301")) == 1
def test_prune_crash_rerun_repairs_pruned_dangling_prev_and_refuses_unexplained( observer_cli_env, monkeypatch: pytest.MonkeyPatch,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="160000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="160000_301", seq=2, prev_segment="160000_300", ) _write_prune_segment( observer_cli_env.journal, segment="160500_300", seq=3, prev_segment="160000_301", audio=b"after", ) _append_upload(prefix, "160000_300") _append_upload(prefix, "160000_301") _write_stream_state(observer_cli_env.journal, last_segment="160500_300", seq=3)
from solstone.apps.observer import prune
def fail_repair(*_args, **_kwargs): raise RuntimeError("interrupt after delete")
with monkeypatch.context() as patch: patch.setattr( prune, "repair_crash_leftovers", lambda *_args, **_kwargs: ([], 0) ) patch.setattr(prune, "repair_stream_chain", fail_repair) with pytest.raises(RuntimeError, match="interrupt after delete"): run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True) assert not ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "160000_301" ).exists() assert any( row.get("type") == "pruned" and row.get("segment") == "160000_301" for row in load_history(prefix, PRUNE_DAY) )
result = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True) assert result.refusals == [] assert result.crash_repaired == 1 assert result.chain_repaired == 0 output = format_result(result) assert "chain-repaired: 0" in output assert "crash-repaired: 1" in output assert ( read_segment_stream( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "160500_300" )["prev_segment"] == "160000_300" )
unexplained = ( observer_cli_env.journal / "chronicle" / PRUNE_DAY / PRUNE_STREAM / "160000_300" ) __import__("shutil").rmtree(unexplained) bad = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True) assert any(refusal.gate == "chain-repair" for refusal in bad.refusals)
def test_prune_execute_twice_is_clean_noop_second_run( observer_cli_env,) -> None: observer = _observer_for_stream() assert save_observer(observer) prefix = observer["key"][:8] _write_prune_segment( observer_cli_env.journal, segment="165000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="165000_301", seq=2, prev_segment="165000_300", ) _write_prune_segment( observer_cli_env.journal, segment="165500_300", seq=3, prev_segment="165000_301", audio=b"after", ) _append_upload(prefix, "165000_300") _append_upload(prefix, "165000_301") _write_stream_state(observer_cli_env.journal, last_segment="165500_300", seq=3)
first = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True) assert result_exit_code(first) == 0 assert [candidate.analysis.segment for candidate in first.deleted] == ["165000_301"] state_after_first = get_stream_state(PRUNE_STREAM) markers_after_first = _stream_marker_snapshot(observer_cli_env.journal) pruned_after_first = _pruned_history(prefix, "165000_301")
second = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True)
assert result_exit_code(second) == 0 assert second.refusals == [] assert second.deleted == [] assert _pruned_history(prefix, "165000_301") == pruned_after_first assert _stream_marker_snapshot(observer_cli_env.journal) == markers_after_first assert get_stream_state(PRUNE_STREAM) == state_after_first
def test_prune_observer_attribution_refuses_no_owner_and_ambiguous_owner( observer_cli_env,) -> None: _write_prune_segment( observer_cli_env.journal, segment="170000_300", seq=1, prev_segment=None, ) _write_prune_segment( observer_cli_env.journal, segment="170000_301", seq=2, prev_segment="170000_300", ) _write_stream_state(observer_cli_env.journal, last_segment="170000_301", seq=2)
no_owner = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True) assert any(refusal.gate == "observer-attribution" for refusal in no_owner.refusals)
first = _observer_for_stream(key="field-one-key") second = _observer_for_stream(key="field-two-key") assert save_observer(first) assert save_observer(second) ambiguous = run_prune(days=[PRUNE_DAY], stream=PRUNE_STREAM, execute=True) assert any( "multiple active observers" in refusal.resolution for refusal in ambiguous.refusals )