vrr/ouroboros
2
0
Fork 0
mirror of https://github.com/razzant/ouroboros.git synced 2026-10-23 23:29:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Merge pull request #1929 from razzant/contrib/maintenance-six-20261015

Browse source 
perf: select maintenance candidates before heavy reads; correct four texts
This commit is contained in:
Ouroboros 2026-10-16 18:45:15 +03:00 • committed by GitHub
commit 96fc9c3ac4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 901 additions and 91 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/CHECKLISTS.md
View file

@ -165,7 +165,7 @@ diff, but it is not loaded as a standalone checklist the way the Change Review
| 10 | Changing `ouroboros/tools/commit_gate.py`? | Coupled surfaces that MUST be updated atomically in the same commit: (a) `tools/preflight_review.py` tool descriptions for `preflight_review` and `review_status` and its `_next_step` strings; (b) the `commit_reviewed` schema in `tools/git.py` (`preflight_reviewer`, `skip_advisory_review`, `reviewers`/`reason` through `compose_commit_panel`) and the preflight-and-tests gate it drives; (c) `docs/DEVELOPMENT.md` Review & Commit Protocol section; (d) the `prompts/SYSTEM.md` Self-Modification section IF the commit-gate rule it states changed. Missing any one has blocked review. |
| 11 | Changing VERSION + pyproject.toml? | Ordering matters: (1) write `VERSION`, `pyproject.toml`, `uv.lock`, `web/package.json` and `web/modules/api_types.js` first; (2) then write the `README.md` badge + changelog row + download links, both install pages' download links, and the `docs/ARCHITECTURE.md` header; (3) then run `pytest`. Never interleave — updating README before VERSION means `test_version_in_readme` will catch a stale badge. |
| 12 | Writing or editing any JS file under `web/modules/`? | New or changed static inline visual properties are blocked: inspect the diff for added/changed `style=""` markup and `.style.<property>` assignments, and use CSS classes/tokens plus `classList`/`hidden` instead. Unchanged legacy hits are debt, not a blocker. A dynamic measured value, or a renderer's count of what it rendered, may update a narrowly named CSS custom property when that is the actual runtime data flow. |
| 13 | Changing LLM output-token budgets? | Grep the whole repo for `max_tokens`, `max_completion_tokens`, `_MAX_TOKENS`, and `max_toks`. Keep `docs/ARCHITECTURE.md` §LLM output token budgets and `tests/test_max_tokens_constants.py` in sync so main-loop, VLM, summaries, compaction, skill publish, and consciousness floors cannot drift independently. |
| 13 | Changing LLM output-token budgets? | Grep the whole repo for `max_tokens`, `max_completion_tokens`, `_MAX_TOKENS`, and `max_toks`. Keep [LLM output token budgets](architecture/07-configuration.md#llm-output-token-budgets) and `tests/test_max_tokens_constants.py` in sync so main-loop, VLM, summaries, compaction, skill publish, and consciousness floors cannot drift independently. |
| 14 | Changing extension loader/dispatch or isolated deps? | Native-risk extension imports and tool/route/WS handlers must stay out-of-process. Add or run regression tests where a native-risk plugin aborts during import and the host survives, plus tool/route child-dispatch tests. Do not "fix" failures by importing native-risk plugin code in `server.py`. |
| 15 | Changing `supervisor/git_ops.py`, `launcher.py`, `server.py`, `ouroboros/tools/review_helpers.py`, `ouroboros/tools/git.py`, tests, or evolution scheduling/checkpoint code? | Prove two invariants before review spend: (1) pytest/preflight cannot mutate the live repo or live `data/` (`OUROBOROS_DATA_DIR` / `OUROBOROS_SETTINGS_PATH` must be isolated, and `OUROBOROS_MANAGED_BY_LAUNCHER` must not leak into test subprocesses); (2) autonomous restart/reset cannot erase active evolution work — it must either land a reviewed local commit or preserve a rescue/transaction recovery pointer and pause/stop the campaign. |
| 16 | Changing `devtools/`? | Keep operator tools isolated from runtime and package discovery: no runtime, web, or release imports from `devtools/`; no generated outputs under `repo/` or live `data/`; no secrets printed or committed. A touched devtool is reviewable executable code even when unrelated devtools stay compact in broad review packs. Domain-specific launch, routing, and methodology guidance stays with the relevant devtool. |

4
docs/PERSISTENCE.md
View file

@ -105,7 +105,7 @@ row here, count-anchored both ways.
| Path | Writer | schema/record marker | Retention | Reset |
|---|---|---|---|---|
| `state/usage.sqlite` (+ `state/usage_attempts.lock` on name-tier installs) | `ouroboros/usage_store.py` single owner (one short SQLite transaction per write: the attempt's one row, its summaries, binding, dirty owners, marker) | `meta.schema_version` 1 plus the import provenance; the lock tier is in `meta.lock_tier` and the header `application_id` | one row per physical attempt, summaries per scope/key; no compaction (docs/USAGE_STORE.md) | monetary history destroyed, budget fences read $0 — unrecoverable unless `usage_attempts.jsonl.imported` is put back as the journal before the next start (it is then imported again); a damaged file is refused typed, never replaced |
| `state/current.sqlite` (+ name-tier `state/current.lock`, `state/current_import.lock`; unpublished .import-pid-token and retired .exported-UTC siblings) | `ouroboros/store.py`: task merges commit body/flags/handoffs in `BEGIN IMMEDIATE`; custody commits rows/outbox. `task_store.read_document` snapshots body+handoffs, closes, then decodes; supplied transactions keep `Txn.load` for read/modify/CAS. Signature/lineage/status/fact selects stay indexed. `store_import.ensure_current_store` builds a sibling and publishes by rename on first write/boot (`store_switch.start`); import makes task-create/Cancel/Pause/Presence and task reads answer 503 `migrating`, supervisor/startup wait for `require_published`, failure keeps the window open. Schema upgrades use the import owner: boot `_ensure`, or `upgrade_current_schema` before a first-touch task mutator/collector on an older present child store; no DDL in passive reads, no history re-import at first touch | `tasks` holds typed facts, obligation flags, `write_seq`, `drive_root`, and nullable `review_collection_pending`/`plan_collection_pending`; `task_bodies` holds JSON; `obligations` holds `custody_open` pending invocations, `custody_refused`, per-task `custody_marks`, `delegated_runs` closing receipts and other owner sets. `task_handoffs` holds `launch_handoffs`/`model_waits`/`retired_model_consumers`: writes change one operation row and task `write_seq`, not the body; reads fill empty body keys; joined local Future tools need no eager row, unbound/legacy contracts stay. `custody_runs` is delegated-run authority (`delegate_custody._apply` inside `emit`), held/closed, indexed by task/project/invocation; `custody_outbox` spans COMMIT→chain append. `meta` holds schema 4 (2 added owner questions, 3 review selection, 4 plan selection), atomic `commit_seq`, boot-reprobed `lock_tier`, published/exporting/retired phase, import provenance/content baselines and export manifest committed before export writes; lifecycle generation 8 is stamped in `state/migrations.json` after publication. Both selectors are derived, outside `FLAG_COLUMNS`/packer duties: NULL unclassified, 0 no candidate, 1 possible collection work. Writers/import/reconcile project both with the body. A schema-3 writer retains its known acceptance selector but omits the plan column: `INSERT OR REPLACE` returns that column to NULL. Both collectors share bounded keyset selection of 1 then NULL through separate partial `(selector, task_id)` indexes; coherent legacy body+`write_seq` snapshots decode off-lock, then batched still-NULL CAS updates only matching sequences. Initial classification reads bodies and commits batches; unchanged classified rows need no repeat, while interruptions/old-writer resets retain that cost. It changes `commit_seq`, not body/`write_seq`; interrupted rows remain NULL. Selector upgrades add no boot-time body-classification pass. Existing collectors retain authority: `plan_record`/`plan_ask` own plan truth/collection, and the late-review owner rechecks task terminality and publishes replies; malformed shapes stay candidates, closed legacy acceptance panels gain no notice without an owed matching receipt | one row per task plus sets; rollback journal, `synchronous=FULL`, no WAL, connection per transaction. DDL/version commit together; prior obligation flags are derived before any first-touch mutation; review/plan classification stays lazy. Boot also derives NULL obligation flags left by older writers. Export before a storeless switch clears the stamp then retires the store as `.exported-<UTC>`; damaged files refuse typed (`StoreError`), never replace | absent: next import rebuilds from `task_results/*.json`; an exported sibling makes readers raise `StoreRetired`. Absent beside the completed generation stamp is LOST, never silently rebuilt; pending switches refuse |
| `state/current.sqlite` (+ name-tier `state/current.lock`, `state/current_import.lock`; unpublished .import-pid-token and retired .exported-UTC siblings) | `ouroboros/store.py`: task merges commit body/flags/handoffs in `BEGIN IMMEDIATE`; custody commits rows/outbox. `task_store.read_document` snapshots body+handoffs, closes, then decodes; supplied transactions keep `Txn.load` for read/modify/CAS. Signature/lineage/status/fact selects stay indexed. `store_import.ensure_current_store` builds a sibling and publishes by rename on first write/boot (`store_switch.start`); import makes task-create/Cancel/Pause/Presence and task reads answer 503 `migrating`, supervisor/startup wait for `require_published`, failure keeps the window open. Schema upgrades use the import owner: boot `_ensure`, or `upgrade_current_schema` before a first-touch task mutator/collector on an older present child store; no DDL in passive reads, no history re-import at first touch | `tasks` holds typed facts, obligation flags, `write_seq`, `drive_root`, and nullable `review_collection_pending`/`plan_collection_pending`/`drive_retention_at`; `task_bodies` holds JSON; `obligations` holds `custody_open` pending invocations, `custody_refused`, per-task `custody_marks`, `delegated_runs` closing receipts and other owner sets. `task_handoffs` holds `launch_handoffs`/`model_waits`/`retired_model_consumers`: writes change one operation row and task `write_seq`, not the body; reads fill empty body keys; joined local Future tools need no eager row, unbound/legacy contracts stay. `custody_runs` is delegated-run authority (`delegate_custody._apply` inside `emit`), held/closed, indexed by task/project/invocation; `custody_outbox` spans COMMIT→chain append. `meta` holds schema 5 (2 owner questions, 3 review selection, 4 plan selection, 5 drive retention), atomic `commit_seq`, boot-reprobed `lock_tier`, published/exporting/retired phase, import provenance/content baselines and export manifest committed before export writes; lifecycle generation 8 is stamped in `state/migrations.json` after publication. Review/plan selectors are derived, outside `FLAG_COLUMNS`/packer duties: NULL unclassified, 0 no candidate, 1 possible collection work. Writers/import/reconcile project both with the body. A schema-3 writer retains its known acceptance selector but omits the plan column: `INSERT OR REPLACE` returns that column to NULL. Both collectors share bounded keyset selection of 1 then NULL through separate partial `(selector, task_id)` indexes; coherent legacy body+`write_seq` snapshots decode off-lock, then batched still-NULL CAS updates only matching sequences. Initial classification reads bodies and commits batches; unchanged classified rows need no repeat, while interruptions/old-writer resets retain that cost. It changes `commit_seq`, not body/`write_seq`; interrupted rows remain NULL. Selector upgrades add no boot-time body-classification pass. `drive_retention_at` uses the same lazy sequence-CAS mechanism over bounded drive-name batches: NULL unclassified, empty text no parseable date, otherwise epoch seconds. `retention.task_result_timestamp` owns first-parseable artifact_finalized_at/completed_at/finished_at/ts, naive UTC; directory-mtime fallback is never persisted. Every body writer/import derives it in the same transaction; older REPLACE writers reset omitted columns to NULL. It only skips unsuitable drives; fresh strict body/age and custody checks still decide removal. This avoids repeated large young-result reads without changing retention. Raw corruption of an intentionally unread body is not detected by this prefilter. Existing collectors retain authority: `plan_record`/`plan_ask` own plan truth/collection, and the late-review owner rechecks task terminality and publishes replies; malformed shapes stay candidates, closed legacy acceptance panels gain no notice without an owed matching receipt | one row per task plus sets; rollback journal, `synchronous=FULL`, no WAL, connection per transaction. DDL/version commit together; prior obligation flags are derived before any first-touch mutation; review/plan classification stays lazy. Boot also derives NULL obligation flags left by older writers. Export before a storeless switch clears the stamp then retires the store as `.exported-<UTC>`; damaged files refuse typed (`StoreError`), never replace | absent: next import rebuilds from `task_results/*.json`; an exported sibling makes readers raise `StoreRetired`. Absent beside the completed generation stamp is LOST, never silently rebuilt; pending switches refuse |
| `state/usage_attempts.jsonl` | `scripts/export_usage_journal.py` (the downgrade export, `usage_store.export_journal`, offline with the server stopped); the journal-era releases wrote it | the journal's validated contract (dense `seq`, `kind`, `state` machine, baseline block) — accepted | the one-time import keeps it in place, never read again here (an older release reads it after a rollback); an export sets the kept one aside (`.pre-export-<UTC>`) and writes it anew — the export plan (`meta.export`: the journal's size, sha256 and row count, the retired and set-aside names) is committed in the store BEFORE any file moves, and while it stands every money write is refused typed (`UsageExportInProgress`), so the journal and the retired store cannot disagree; a kept journal whose sha256 is neither the imported nor the exported one stops the export (`move it aside`) and is never merged; an abandonment puts the store back from the name in the mark, restores the set-aside journal and clears the mark last | an exported journal deleted before the downgrade loses the money the older release would read; the retired store (`usage.sqlite.exported-<UTC>`) still holds it |
| `state/usage_attempts.quarantine.jsonl` | `ouroboros/usage_store.py` (the one-time import truncates a torn final journal row in place and moves its bytes here) | one row per quarantined tail, the torn bytes verbatim | written once per repair, never appended otherwise | the store's integrity flag loses its evidence: that money still reads as degraded, never final |
| `state/skill_review_root_tasks.jsonl` (+ `state/skill_review_root_tasks.gaps.jsonl`) | `ouroboros/skill_review_history.py` (`_append_root_task_projection_once`: one compact row per root task appended when its skill review lands; a row the projection could not attribute goes to the `.gaps.jsonl` ledger via `_record_root_task_projection_gap`) | rows carry `usage_attribution_schema: physical_attempt_v1`; no version key — accepted (derived index over the per-skill `review_history.jsonl`, P7) | unbounded append; reads are BOUNDED (`skill_readiness.py`: 1 MiB / 512 records tail, a truncated or gapped read is disclosed as `projection_incomplete`) and the startup hot-store check warns past `SKILL_REVIEW_ROOT_TASKS_WARN_BYTES` | delete with `state/`; the index is not rebuilt — readiness and the acceptance packet read `projection_incomplete` until new rows accrue (the per-skill histories keep the truth) |
@ -217,7 +217,7 @@ row here, count-anchored both ways.
| `archive/**` (rotated segments, `rescue/`, `usage_import/`, `managed_repo/`) | rotation + `supervisor/git_ops_rescue.py`, `usage_journal.py`, `launcher_bootstrap.py` | segments inherit source shape; usage_import carries sha256 sidecar | UNBOUNDED BY DESIGN — durable history, never GC'd (P1) — accepted | memory horizon truncated; rescue copies of uncommitted work destroyed |
| `archive/usage_ledger/segment_*.jsonl` → `segment_ep*.jsonl.xz` (+ `segment_ep*.jsonl.xz.json` size/sha256 sidecar) | none since the usage store: the exact pre-compaction journal bytes the retired compactor wrote; `evidence_packs.archive_usage_segments` (from the packer child) re-encodes each `segment_ep*.jsonl` as one xz stream (preset 1), verifies the decoded bytes against the recorded sha256, then unlinks the original | each segment is a whole journal generation; the sidecar records the original's name, size and sha256 | UNBOUNDED BY DESIGN but no longer growing — retained evidence, read only by the explicit history audit (`ouroboros/model_send_seal.py`, a plain id scan that opens `.xz` by suffix); a loose placement is replaced by a sha-verified compressed copy, the bytes kept forever; `unpack_evidence.py --usage-archive` restores the plain files | the audit reports history it cannot read (missing, damaged or truncated) as unknown, never as orphan seals; money is unaffected |
| `observability/{calls,blobs,salvaged}/**` | `ouroboros/observability.py` (0700/0600, CAS gzip-1; `llm_request` omits canonically equal `send_messages`, restored from `messages` on read); model-send records beside the call manifests: `model_send_seal` block + write-once `<attempt>.model_send_violation.json` typed facts (`ouroboros/model_send_seal.py`) | manifests `schema_version: 1` + custody/redaction markers; blobs sha-verified on read; `model_send_seal.seal_version: 1` with the `canonical_json_v1` basis string | kept indefinitely: a settled task's blobs move into its evidence pack (`observability/packs/`, the row below) and their loose files are unlinked only after the pack is re-verified from disk — a blob a non-settled task still references stays loose; promotion hardlinks immutable CAS (copy on any link error, a gzip-1 copy when the child's bytes are already packed); `OUROBOROS_OBSERVABILITY_RETENTION_DAYS` is retired | `result_ref`/`manifest_ref` dangle (strict readers raise); missing delegated bodies refuse replay as unknown, while custody still blocks replacement and protects snapshots; salvaged outputs lost; missing seals report `unlogged_attempt` at explicit audit |
| `observability/packs/<task>[.<gen>].xz` + `<task>[.<gen>].members.json`, `observability/packs/index.sqlite`, `observability/packs/unpack_evidence.py` + `README.md`, scratch `observability/packs/.publications` (locks: `state/packs_index.lock`, `state/packs_publication.lock`, `state/packs_sweep.lock`) | `ouroboros/evidence_packs.py`, `ouroboros/evidence_packer.py` child; `storage_pack` in `logs/supervisor.jsonl`; `blob_bytes`/`handle_bytes` resolve loose → index → pack; `restore_checkpoint_handles` restores descendant prefixes at switches | write-once sidecar `format_version: 1`: header {task_id, generation, codec `xz`, preset 1, dict_size, raw_total, packed_total, pack_sha256, created_at}, members [{sha256, kind, size, offset, path}]; checkpoints first, then blobs smallest first by (size, sha256) (older packs: manifest first-appearance order; readers follow the sidecar offsets); every member complete in its own pack. Index schema 2, key sha/kind/task/generation, DELETE/FULL, `OUPE`/`OUPN`: rebuilt from sidecars if missing, bad `quick_check` or wrong version; separate from `state/current.sqlite` | temp + fsync → pack → sidecar LAST; disk-verify every member before unlink; unpacker/README first; generations never reused, only new members packed, bytes kept forever. Task-local stamps; an examined task is recorded swept unless busy or deferred (a blob kept for a live sharer leaves with the sharer's own pass); one money query/outbox read per pass/tick. One live-reference census outside locks, then only new publication refs under the publication lock; scratch journal reset each pass, removed on exit. Before each unlink the census is refreshed outside the locks: census-settled tasks whose row, duties, money or delivery moved are re-read, and one that left settlement keeps its refs; activity elsewhere never stops a pass. Inside writer/publication locks only an O(1) staleness check (stale defers that candidate), its recheck, journal suffix and unlinks, in `UNLINK_SECTION_SEC` sections under a rollback barrier with a waiter's gap between them (2N settlement checks per quiet one-section backfill); a manifest waits up to `PUBLICATION_WAIT_SEC` for the publication lock. No loose members: no pack verification | missing pack: `FileNotFoundError`; corrupt member/sidecar: `evidence_corrupt`; bad sidecars refuse restore/preflight, neighbours stay readable. Standalone unpacker restores a task, `--all` or `--usage-archive`; scratch is disposable |
| `observability/packs/<task>[.<gen>].xz` + `<task>[.<gen>].members.json`, `observability/packs/index.sqlite`, `observability/packs/unpack_evidence.py` + `README.md`, scratch `observability/packs/.publications` (locks: `state/packs_index.lock`, `state/packs_publication.lock`, `state/packs_sweep.lock`) | `ouroboros/evidence_packs.py`, `ouroboros/evidence_packer.py` child; `storage_pack` in `logs/supervisor.jsonl`; `blob_bytes`/`handle_bytes` resolve loose → index → pack; `restore_checkpoint_handles` restores descendant prefixes at switches | write-once sidecar `format_version: 1`: header {task_id, generation, codec `xz`, preset 1, dict_size, raw_total, packed_total, pack_sha256, created_at}, members [{sha256, kind, size, offset, path}]; checkpoints first, then blobs smallest first by (size, sha256) (older packs: manifest first-appearance order; readers follow the sidecar offsets); every member complete in its own pack. Index schema 2, key sha/kind/task/generation, DELETE/FULL, `OUPE`/`OUPN`: rebuilt from sidecars if missing, bad `quick_check` or wrong version; separate from `state/current.sqlite` | temp + fsync → pack → sidecar LAST; disk-verify every member before unlink; unpacker/README first; generations never reused, only new members packed, bytes kept forever. Task-local stamps; an examined task is recorded swept unless busy or deferred (a blob kept for a live sharer leaves with the sharer's own pass); one money query/outbox read per discovery/pass. `tick` only reserves the existing thread; archive-first discovery stops after the first changed stamp, polls Stop between entries and takes nonwaiting store/index reads without cache rebuild. Quiet nonempty roots still scan, spawning no child; contention defers. The child reconstructs invalid indexes and owns a fresh census, never the parent's hint. The live discovery thread remains in restart custody. One live-reference census outside locks, then only new publication refs under the publication lock; scratch journal reset each pass, removed on exit. Before each unlink the census is refreshed outside the locks: census-settled tasks whose row, duties, money or delivery moved are re-read, and one that left settlement keeps its refs; activity elsewhere never stops a pass. Inside writer/publication locks only an O(1) staleness check (stale defers that candidate), its recheck, journal suffix and unlinks, in `UNLINK_SECTION_SEC` sections under a rollback barrier with a waiter's gap between them (2N settlement checks per quiet one-section backfill); a manifest waits up to `PUBLICATION_WAIT_SEC` for the publication lock. No loose members: no pack verification | missing pack: `FileNotFoundError`; corrupt member/sidecar: `evidence_corrupt`; bad sidecars refuse restore/preflight, neighbours stay readable. Standalone unpacker restores a task, `--all` or `--usage-archive`; scratch is disposable |
| `claudexor/**` | EXTERNAL writer — claudexord; Ouroboros owns its lifecycle, `daemon.log` and `ouroboros-owned.json` marker | marker unversioned; engine owns SQL schemas and the semantic reader floor | engine-owned state, retention and recovery evidence; never host GC; migration and recovery differ as described below | engine state, retained evidence and harness logins/profiles lost; deleting the root is not a cache reset |
| `playwright-browsers/` | `ouroboros/tools/browser.py` (vendor install) | none — vendor tree | no GC — accepted (vendor cache) | re-downloaded on next browser use |
| `cache/pip`, `cache/uv` | `ouroboros/test_environment.py` | package-manager formats; TEST ROOT ONLY, not live data | disposable verification-root lifetime | recreated by explicit dependency setup |

2
docs/architecture/01-high-level-architecture.md
View file

@ -243,7 +243,7 @@ ouroboros/ — agent core and shared runtime (§6)
body_candidate.py, body_adoption.py, body_switch.py — Own-body worktree authoring and restart-bound adoption of an exact reviewed commit before body imports (§6 Own-body candidates; §2)
artifacts.py — attachment staging into `artifact_store/attachments/`, artifact records, scratch fingerprints (`.scratch_manifest.json`), the undeclared-output guard, partial tool-evidence reads
chat_uploads.py, confined_files.py — the owner-attachment store (`data/uploads`) with byte-proven media kind; directory-confined regular-file open (DESIGN "Chat attachments")
retention.py — GC retention SSOT: clamp and age cutoff
retention.py — GC clamp, cutoff and exact task-drive retention date (§6)
workspace_preflight.py — read-only external-workspace snapshot for gateway task creation
project_sources.py — folder attach validation (realpath, not the home root, no repo/data overlap); opt-in `init_git`; a server-side clone never prompts and reports a typed `auth_required`; attaching is the trust grant (`trusted_at`)
promotion_source.py — promoted-task source admission after an executor/id reservation

4
docs/architecture/05-supervisor-loop.md
View file

@ -80,7 +80,9 @@ Surplus retirement and cap-full warm Resume share a quiescence token probe; a wo
### Maintenance and custody
`server_maintenance.py` gives schedules, cancel, delivery, pause-notice, usage, post-task recovery, custody and reconcile their own single-flight latches; domain claims/receipts arbitrate interleavings, not pass order. `bind_body_lifespan` starts the body after store publication; replacement joins closing physical jobs off-loop; cold holders enter orphan-exclusion before recovery (cadence: §6); generation closure or a returning stop's named queue admits nothing and claims no physical exit (§10 invariant 24). Sweeps remember a deterministic refusal until its source changes: a child-ref refusal (`invalid_source`) is walked again only once a watched file, its rows or the generation change (`source_retention._unavailable_basis` docstring); cost projection takes a busy store as no observation and proves absence in one statement; the delegated-run sweep leaves review-owned pending invocations to their panel (`delegate_terminal` docstring) and captures a finished run without a file work product as no changes only after success, otherwise as nothing captured (`delegate_directory.capture_directory_result`). Process reaping checks PID, birth, command, task and generation, never command-line classes or another install. The phase watchdog journals loop stalls and reports wedged actors with a restart hint; neither kills a thread.
`server_maintenance.py` gives schedules, cancel, delivery, pause-notice, usage, post-task recovery, custody and reconcile separate single-flight latches; domain claims/receipts, not pass order, arbitrate. `bind_body_lifespan` starts after store publication; replacement joins closing jobs off-loop; cold holders enter orphan-exclusion before recovery (§6). Generation closure/returning Stop admits nothing and proves no exit (§10.24). An `invalid_source` refusal is retried only when its watched file, rows or generation change (`source_retention._unavailable_basis`); cost reads treat busy as no observation and prove absence in one statement. Delegated sweeps leave review-owned invocations to panels (`delegate_terminal`); no-file captures mean no changes only after success, otherwise nothing captured (`delegate_directory.capture_directory_result`). Reaping checks PID/birth/command/task/generation, never command classes or another install. Watchdogs journal stalls, report wedged actors and suggest Restart; they kill no thread.
Packer `tick` reserves its existing thread: stop-aware, archive-first discovery exits at the first changed stamp. Contention defers; quiet nonempty roots scan without spawning. Discovery stays in the exit census, releasing reconcile without hiding readers. The child retains its own fresh `PackSweep` baseline/refresh/journal and unlink authority.
Cooperative project checkpointing (`supervisor/events_coop_checkpoint.py`) runs on two equivalent quiescence triggers, a root settling without live descendants and the last child settling under an already-terminal root, because a root-scope budget stop terminalizes the root before its children (its handler docstrings). The bounded git chain revalidates quiescence under Q just before mutation. Only host-minted project roots qualify: owner-attached folders are never auto-committed, credential-shaped files stay excluded and disclosed, and each material success, skip or error gets a durable receipt.

4
docs/architecture/06-agent-core.md
View file

@ -38,7 +38,9 @@ Default `pending_review=wait` keeps the same operation in every mode through mai
#### Headless finalization and workspace patch capture
A workspace task's completion diffs against the captured preflight base (`task_constraint.base_sha`; task-local commits stay in the delta, not `git diff HEAD`); a moved HEAD fails closed only for `self_worktree` (a shared tree relies on reverse-patch verification); an unborn repository diffs against the canonical empty tree. `workspace_patch_capture.py` streams the tracked binary diff plus admitted untracked files under `workspace_patch_rules.py`'s pure rules, each vetoed entry excluded with its reason; oversized or binary untracked outputs ride a manifest+zip artifact, not a patch. Generated output follows the project's `.gitignore`, not a host directory-name veto, so build deliverables are kept; a sensitive-looking untracked credential is excluded per file (`sensitive_blocked`). `workspace_patch.json` is written for every workspace finalization, no-change and failed included; `workspace.patch` only for `ready_with_changes`; the CLI strict-patch contract reads them (§1). Forked/empty execution drives live at `data/state/headless_tasks/<task_id>/data`; results return to the canonical root, artifacts to `data/task_results/artifacts/<task_id>/`. Late copy-back preserves terminal results; terminal file adoption drops child `state/code_intel`. `task_custody.settle_child_drive` decides every child-drive deletion: the drive goes only when the supervisor's probe proves no live owner for every occupant, the durable row is settled and the canonical row holds every recorded obligation; missing or mismatched material keeps it until a restore converges.
Workspace completion diffs against preflight `task_constraint.base_sha`, including local commits; unborn repos use Git's empty tree. A moved HEAD refuses only `self_worktree`; shared trees use reverse-patch verification. `workspace_patch_capture.py` streams tracked binary changes and admitted untracked files under `workspace_patch_rules.py`, disclosing exclusions. Oversized/binary untracked outputs use manifest+ZIP. Project `.gitignore`, not host directory names, governs generated output; credentials are excluded per file (`sensitive_blocked`). Every finalization writes `workspace_patch.json`, including no-change/failure; only `ready_with_changes` writes `workspace.patch` (CLI: §1). Forked/empty drives use `data/state/headless_tasks/<task_id>/data`; results return to the canonical root, artifacts to `data/task_results/artifacts/<task_id>/`. Late copy-back preserves terminal results; adoption drops child `state/code_intel`.
`headless` selects drive names in bounded batches using `task_store.drive_retention_facts`: the writer's exact effective date avoids rereading young bodies; legacy NULLs classify lazily, directory mtime stays layout-local. Fresh strict body/age checks precede `task_custody.settle_child_drive`, which alone deletes: the supervisor must prove every occupant inactive, the row settled and every obligation canonically held. Missing/mismatched material stays until restoration converges. Projection and first-parseable date semantics: `retention.py`, PERSISTENCE.
#### Owner routing verbs

4
docs/architecture/11-frozen-contracts-v1.md
View file

@ -45,7 +45,7 @@ ABI promise: the frozen shapes and helper semantics skills and browser consumers
### 11.2 What is NOT frozen (intentionally)
The full `ToolContext` dataclass stays mutable implementation detail; the protocol pins only the minimum. Only shape keys are pinned, not raw WebSocket/HTTP values, and only the frontmatter schema, not the free-form SKILL.md body. `state/state.json` and `queue_snapshot.json` carry no `_schema_version` and read as version 0; `task_results/*.json` uses stamped admission owned by `task_result_schema.py` (§11.4).
`ToolContext` stays mutable; its protocol pins only the minimum. Shape keys, not raw WebSocket/HTTP values, and frontmatter, not the SKILL.md body, are pinned. State and queue writers stamp `_schema_version: 1`; readers tolerate unstamped legacy version 0. Task documents in `current.sqlite` use `task_result_schema.py` admission (§11.4).
### 11.3 What to do when extending
@ -59,4 +59,4 @@ Add the field to its active frozen owner (§11.1), keep existing consumers worki
- Review pool: `REVIEW_POOL_MIGRATED_SETTING_KEYS` (a `RETIRED_SETTING_KEYS` subset) migrates `OUROBOROS_REVIEWER_SLOTS` and §7's listed surface keys into `OUROBOROS_SUBAGENTS` reviewer rows. Only disk settings migrate; unconvertible input stays intact with its error. First save preserves `state/review_migrations/` and `state.json:review_pool_migrations`; first owner boot reports it. One list serves review (`GET /api/review-pool`); owners, factory fallback, receipts and ignored-environment notice: §7 Reading and writing the settings document and Review pool.
- Gateway envelopes: the `api_v1` re-export and the `cost_usd`/`cost_usd_with_children`, `telegram_chat_id`, `project_last_viewed`/`project_hidden` aliases are gone; stored rows still read tolerantly, and `UiPreferencesResponse` answers 400 on an unknown key.
- Plugin ABI: an existing hash-bound PASS keeps loading; manifest negotiation and defaults: §11.1 (`PluginAPI`).
- Durable task rows: every `task_results/<id>.json` write stamps `_schema_version: 1`, with no converter for older rows; an inadmissible row is quarantined (log-only) and keeps its id occupied.
- Durable task rows: writes stamp `_schema_version: 1`; `task_results/<id>.json` is import/export, not current authority. Inadmissible rows are quarantined (log-only), never converted, and keep their ids occupied.

2
docs/inventories/DATA_LAYOUT_INVENTORY.md
View file

@ -2,7 +2,7 @@
Machine extraction of the `docs/ARCHITECTURE.md` "Data layout (`~/Ouroboros/`)" tree — the durable-file orientation carrier (this tree's counterpart of the reference PERSISTENCE_OWNERS derivation checklist) — regenerated by `python scripts/regenerate_inventories.py`. Do not edit. Every entry is probed against reality: repo entries must exist as tracked paths; data-plane entries must appear as a literal in the runtime sources that construct them. A durable file renamed or removed in code while its tree row survives = red (`tests/test_generated_inventories.py`).
Source: `docs/architecture/01-high-level-architecture.md`, physical LF lines 592-696; UTF-8 SHA-256 `7774893a699ffe270ef44dfc32f40ceaab34d91a7badbbeeb9751a9e8671b409`.
Source: `docs/architecture/01-high-level-architecture.md`, physical LF lines 592-696; UTF-8 SHA-256 `105b6a802cedcbff3f9f77a0af7c1cd93c74b978b9f0e0652d0c3531e6369a44`.
- entries: **94** (code-ref: 87, repo-dir: 6, repo-path: 1)

2
docs/inventories/FROZEN_CONTRACTS_INVENTORY.md
View file

@ -2,7 +2,7 @@
Machine extraction of `docs/ARCHITECTURE.md` §11.1 (the frozen-ABI SSOT), regenerated by `python scripts/regenerate_inventories.py`. Do not edit — edit the owning chapter named in the Source line and regenerate; `tests/test_generated_inventories.py` pins byte-identity and the resolution invariants (a §11.1 row whose owner or anchor file disappeared from the tree = red).
Source: `docs/architecture/11-frozen-contracts-v1.md`, physical LF lines 7-45; UTF-8 SHA-256 `564b86b911437c0d46ee9a16fa88e3af793d18f67ab325b188b725eb2861d428`.
Source: `docs/architecture/11-frozen-contracts-v1.md`, physical LF lines 7-45; UTF-8 SHA-256 `2a501a44ca67557c8a92178984c173870ec987b7e95bf7ac6f7d31d281ec05f1`.
- table rows: **34**
- browser-envelope prose owners:

97
ouroboros/evidence_packer.py
View file

@ -11,7 +11,7 @@ import pathlib
import sqlite3
import threading
import time
from typing import Any, Dict, List, Optional, Tuple
from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple
from ouroboros import evidence_packs as packs, store
from ouroboros.startup_historical_audit import _REPORT_BYTES, finish_maintenance_job, spawn_maintenance_job
@ -52,25 +52,36 @@ def _stamp(root: pathlib.Path, task_id: str, seq, obligation_rows: dict, externa
return json.dumps(stamps, sort_keys=True, separators=(",", ":"))
def _loose_task_ids(root: pathlib.Path) -> set:
def _loose_task_ids(root: pathlib.Path, stop: Optional[Callable[[], bool]] = None) -> set:
ids = set()
with contextlib.suppress(OSError):
ids.update(entry.name for entry in os.scandir(root / "observability" / "calls") if entry.is_dir(follow_symlinks=False))
with contextlib.suppress(OSError):
for entry in os.scandir(root / "task_results" / "artifacts"):
if entry.is_dir(follow_symlinks=False) and (pathlib.Path(entry.path) / "source_handles" / "context_checkpoints").is_dir():
ids.add(entry.name)
for directory, checkpoints in ((root / "observability" / "calls", False),
(root / "task_results" / "artifacts", True)):
if stop is not None and stop():
break
with contextlib.suppress(OSError), os.scandir(directory) as entries:
for entry in entries:
if stop is not None and stop():
return ids
if entry.is_dir(follow_symlinks=False) and (not checkpoints or
(pathlib.Path(entry.path) / "source_handles" / "context_checkpoints").is_dir()):
ids.add(entry.name)
return ids
def swept_stamps(root: pathlib.Path) -> Optional[Dict[str, str]]:
def swept_stamps(root: pathlib.Path, *, discovery: bool = False) -> Optional[Dict[str, str]]:
"""Per task, the stamps the last sweep recorded; ``None`` when the cache is unavailable."""
if not packs.packs_dir(root).is_dir():
return {}
try:
with packs.open_index(root) as conn:
with packs.open_index(root, **({"wait_sec": 0, "rebuild": False} if discovery else {})) as conn:
return {row["task_id"]: row["stamps"] for row in conn.execute("SELECT task_id, stamps FROM swept")}
except (store.StoreError, sqlite3.Error, OSError):
except store.StoreBusy:
if discovery:
raise # advisory discovery skips this pass instead of waiting or spawning on contention
return None
except (store.StoreError, sqlite3.Error, OSError) as exc:
if discovery and isinstance(exc, sqlite3.OperationalError) and store.busy(exc):
raise store.StoreBusy("evidence index read busy", reason="contention") from exc
return None
@ -80,22 +91,50 @@ def record_swept(root: pathlib.Path, task_id: str, stamps: str) -> None:
conn.execute("INSERT OR REPLACE INTO swept VALUES (?,?)", (task_id, stamps))
def changed_stamps(root: pathlib.Path, *, obligation_rows: Optional[dict] = None, external: Optional[dict] = None,
seqs: Optional[dict] = None) -> Dict[str, str]:
def _changed_stamps(root: pathlib.Path, *, obligation_rows: Optional[dict] = None, external: Optional[dict] = None,
seqs: Optional[dict] = None, stop: Optional[Callable[[], bool]] = None) -> Iterator[Tuple[str, str]]:
"""Tasks with loose evidence whose directories or result changed since a sweep recorded them, with their
stamps. The rows and duties come from ONE short read that is closed before the directory probes."""
root = pathlib.Path(root)
ids = _loose_task_ids(root)
if not ids:
return {}
ids = _loose_task_ids(root) if stop is None else _loose_task_ids(root, stop)
if not ids or (stop is not None and stop()):
return
if obligation_rows is None or seqs is None:
with store.open_store(root, write=False) as txn:
obligation_rows = packs.settlement_obligations(root, txn=txn) if obligation_rows is None else obligation_rows
seqs = txn.write_seqs() if seqs is None else seqs
external = packs.settlement_external(root) if external is None else external
swept = swept_stamps(root)
stamps = {task: _stamp(root, task, seqs.get(task), obligation_rows, external) for task in sorted(ids)}
return {task: stamp for task, stamp in stamps.items() if swept is None or swept.get(task) != stamp}
if stop is not None and stop():
return
external = packs.settlement_external(root, **({"timeout_sec": 0} if stop is not None else {})) if external is None else external
if stop is not None and stop():
return
swept = swept_stamps(root, discovery=stop is not None)
for task in sorted(ids):
if stop is not None and stop():
return
stamp = _stamp(root, task, seqs.get(task), obligation_rows, external)
if swept is None or swept.get(task) != stamp:
yield task, stamp
def changed_stamps(root: pathlib.Path, **census) -> Dict[str, str]:
"""Full child census; never reuse the parent's advisory discovery as unlink authority."""
return dict(_changed_stamps(root, **census))
def packing_pending(root: pathlib.Path, stop: Callable[[], bool]) -> bool:
"""Advisory, stop-aware discovery: archives first, then only the first changed stamp.
Contended stores defer the pass. Index reconstruction belongs to the supervised
child, not this reader. A quiet nonempty root still needs a full stamp walk.
"""
if stop():
return False
if archive_pending(root):
return not stop()
with store.acquisition_budget(0):
changed = next(_changed_stamps(root, stop=stop), None)
return changed is not None and not stop()
def changed_tasks(root: pathlib.Path, **census) -> List[str]:
@ -288,7 +327,10 @@ class EvidencePacker:
self._thread: Optional[threading.Thread] = None
def tick(self, data_root: pathlib.Path, repo_dir: pathlib.Path) -> bool:
"""Called from the maintenance sweep; spawns at most one child, never on the first tick."""
"""Reserve one discovery thread, never on the boot tick; no filesystem walk here.
True means discovery started, not that a packing child was needed or spawned.
"""
with self._lock:
if self._stopped.is_set() or (self._thread is not None and self._thread.is_alive()):
return False
@ -296,12 +338,6 @@ class EvidencePacker:
self._armed = True # the boot tick only arms the packer
return False
root = pathlib.Path(data_root)
try:
if not changed_tasks(root) and not archive_pending(root):
return False
except Exception:
log.debug("Evidence pack candidate check failed", exc_info=True)
return False
self._thread = threading.Thread(target=self._run, args=(root, pathlib.Path(repo_dir)),
name="evidence-packer", daemon=True)
self._thread.start()
@ -328,7 +364,12 @@ class EvidencePacker:
def _run(self, data_root: pathlib.Path, repo_dir: pathlib.Path) -> None:
started = time.monotonic()
try:
if self._stopped.is_set():
try:
pending = packing_pending(data_root, self._stopped.is_set)
except Exception:
log.debug("Evidence pack candidate check failed", exc_info=True)
return
if not pending or self._stopped.is_set():
return
process = spawn_maintenance_job(data_root, repo_dir, "evidence_packer")
fields = finish_maintenance_job(self, process, data_root, started, _report_fields, packs.record_storage_pack)

31
ouroboros/evidence_packs.py
View file

@ -220,7 +220,7 @@ def index_tier(root: pathlib.Path) -> str:
return store.TIER_ENFORCED if kernel_file_locks_enforced(root / "state" / store.LOCK_NAME) else store.TIER_NAME
def _index_valid(path: pathlib.Path, tier: str) -> bool:
def _index_valid(path: pathlib.Path, tier: str, *, wait_sec: float = INDEX_WAIT_SEC) -> bool:
identity = store.file_identity(path)
if identity is None:
return False
@ -234,9 +234,13 @@ def _index_valid(path: pathlib.Path, tier: str) -> bool:
if store.header_tier(path, _APPLICATION_ID) != tier:
return False
try:
with contextlib.closing(store.connect(path, tier, wait_sec=INDEX_WAIT_SEC)) as conn:
with contextlib.closing(store.connect(path, tier, wait_sec=wait_sec)) as conn:
ok = conn.execute("PRAGMA quick_check").fetchone()[0] == "ok"
version = conn.execute("SELECT value FROM meta WHERE key='schema_version'").fetchone()
except sqlite3.OperationalError as exc:
if store.busy(exc):
raise store.StoreBusy("evidence index validation busy", reason="contention") from exc
return False
except sqlite3.Error:
return False
valid = ok and version is not None and version[0] == str(INDEX_SCHEMA_VERSION)
@ -278,13 +282,18 @@ def _rebuild_index(root: pathlib.Path, path: pathlib.Path, tier: str) -> None:
store.fsync_directory(path.parent)
@contextlib.contextmanager
def open_index(root: Any, *, write: bool = False, wait_sec: float = INDEX_WAIT_SEC) -> Iterator[sqlite3.Connection]:
def open_index(root: Any, *, write: bool = False, wait_sec: float = INDEX_WAIT_SEC,
rebuild: bool = True) -> Iterator[sqlite3.Connection]:
"""One transaction on the index cache under the name lock; a missing, damaged or
other-version cache is rebuilt from the sidecars first. ``store.StoreBusy`` on contention."""
other-version cache is rebuilt from the sidecars first. ``rebuild=False`` is
advisory discovery: unavailable cache raises instead of walking all sidecars.
``store.StoreBusy`` on contention."""
root = pathlib.Path(root)
path, tier = packs_dir(root) / INDEX_NAME, index_tier(root)
with store.name_lock(root, INDEX_LOCK_NAME, timeout_sec=wait_sec):
if not _index_valid(path, tier):
if not _index_valid(path, tier, wait_sec=wait_sec):
if not rebuild:
raise sqlite3.DatabaseError("evidence index requires reconstruction")
_rebuild_index(root, path, tier)
with contextlib.closing(store.connect(path, tier, wait_sec=wait_sec)) as conn:
with store.transaction(conn, write=write):
@ -670,8 +679,9 @@ def task_settlement_facts(root: Any, task_id: str, *, obligation_rows: Optional[
}
def settlement_external(root: Any) -> dict:
"""One money query and one outbox read for every task in a pass; None stays unknown."""
def settlement_external(root: Any, *, timeout_sec: Optional[float] = None) -> dict:
"""One money query and outbox read per pass. A bounded advisory read propagates
unavailability to defer discovery; ordinary child reads keep unknown as None."""
from ouroboros import usage_store
from ouroboros.usage_ledger import LEDGER_REL
from supervisor.terminal_delivery import pending_deliveries
@ -680,15 +690,20 @@ def settlement_external(root: Any) -> dict:
money = set()
if (root / usage_store.STORE_REL).exists() or (root / LEDGER_REL).exists():
try:
with usage_store.hold(root, write=False, migrate=False) as txn:
with usage_store.hold(root, write=False, migrate=False,
**({"timeout_sec": timeout_sec} if timeout_sec is not None else {})) as txn:
money = {row[0] for row in txn.conn.execute(
"SELECT DISTINCT task_id FROM attempts WHERE kind='attempt' "
"AND state IN ('reserved','dispatched','unresolved')")}
except Exception:
if timeout_sec is not None:
raise
money = None
try:
delivery = {str(event.get("task_id") or "") for event in pending_deliveries(root, strict=True)}
except Exception:
if timeout_sec is not None:
raise
delivery = None
return {"money": money, "delivery": delivery}

59
ouroboros/headless.py
View file

@ -11,7 +11,6 @@ import shutil
import subprocess # noqa: F401
import tempfile # noqa: F401
import threading # noqa: F401
from datetime import datetime, timezone
from hashlib import sha256
from typing import Any, BinaryIO, Dict, Iterable, List, Optional, Sequence, Tuple # noqa: F401
@ -22,6 +21,7 @@ from ouroboros.task_results import (
cancellation_blocks_child_result, load_task_result, validate_task_id, write_task_result,
)
from ouroboros.utils import atomic_write_json, utc_now_iso
from ouroboros.retention import task_result_timestamp as _timestamp_from_result
from ouroboros.headless_status import ( # noqa: F401
ARTIFACT_STATUS_FAILED,
ARTIFACT_STATUS_FINALIZING,
@ -164,21 +164,6 @@ def _resolve_retention_days(retention_days: Optional[int]) -> int:
return retention_days
def _timestamp_from_result(result: Dict[str, Any], fallback: float) -> float:
for key in ("artifact_finalized_at", "completed_at", "finished_at", "ts"):
raw = str(result.get(key) or "").strip()
if not raw:
continue
try:
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return float(parsed.timestamp())
except ValueError:
continue
return fallback
def _effective_task_result(parent: pathlib.Path, task_id: str) -> Dict[str, Any]:
"""Prune reads the projected result; a projection failure reads canonical."""
try:
@ -201,11 +186,31 @@ def _prompt_settlement(result: Dict[str, Any]) -> bool:
and str(result.get("delegation_role") or "") == "subagent")
def _drive_retention_rows(parent: pathlib.Path, names: List[str], stop: Any) -> Iterable[Tuple[str, Any]]:
"""Select only this layout's names in bounded transactions; never hold one across a yield.
A selector/upgrade failure does not become absence: the existing strict reader
handles each name, preserving missing/quarantined/unreadable-result outcomes.
"""
from ouroboros import store
from ouroboros.task_store import DRIVE_RETENTION_BATCH, drive_retention_facts
for start in range(0, len(names), DRIVE_RETENTION_BATCH):
batch = names[start:start + DRIVE_RETENTION_BATCH]
try:
facts = drive_retention_facts(parent, batch, stop=stop)
except (store.StoreError, ValueError, TypeError):
facts = {}
for task_id in batch:
yield task_id, facts.get(task_id)
def _prune_drives(base: pathlib.Path, parent: pathlib.Path, *, drive_of: Any, not_terminal: str,
retention_days: Optional[int], now: Optional[float], live: Any, guard: Any, stop: Any,
budget: Optional[int], after: str, extra_checks: Any) -> Dict[str, Any]:
"""The prune both drive layouts share: candidates from the DURABLE canonical row (a projection is not
custody), settled through ``task_custody.settle_child_drive`` in name order after ``after`` (wrapping), at
"""The prune both layouts share: cheap per-name facts select possible work;
fresh strict canonical status/date and existing checks still decide eligibility.
Settlement through ``task_custody.settle_child_drive`` runs in name order after ``after`` (wrapping), at
most ``budget`` attempts per call; ``cursor`` is the last attempted drive, ``deferred`` the unreached rest."""
from ouroboros.retention import age_cutoff
from ouroboros.task_custody import settle_child_drive
@ -219,11 +224,27 @@ def _prune_drives(base: pathlib.Path, parent: pathlib.Path, *, drive_of: Any, no
names = sorted(entry.name for entry in base.iterdir() if entry.is_dir())
names = [name for name in names if name > after] + [name for name in names if name <= after]
attempts = 0
for task_id in names:
for task_id, fact in _drive_retention_rows(parent, names, stop):
task_dir = base / task_id
report["scanned"] += 1
try:
validate_task_id(task_id)
if fact is not None and fact["body_present"] and fact["drive_retention_at"] is not None:
status = str(fact.get("status") or "").lower()
if status not in _FINAL_STATUSES:
report["skipped"].append({"task_id": task_id, "reason": not_terminal, "status": status})
continue
if not _prompt_settlement(fact):
raw = fact["drive_retention_at"]
timestamp = float(raw) if raw else task_dir.stat().st_mtime
if timestamp > cutoff:
report["skipped"].append({"task_id": task_id, "reason": "younger_than_retention"})
continue
if stop is not None and stop():
report["deferred"].append(task_id)
continue
# A hint never authorizes removal. In particular a date changed after
# selection must be checked here; settlement rechecks custody, not age.
result = load_task_result(parent, task_id, strict=True) or {}
status = str(result.get("status") or "").lower()
if status not in _FINAL_STATUSES:

40
ouroboros/retention.py
View file

@ -18,7 +18,8 @@ migration in ``config.py``).
from __future__ import annotations
import time
from typing import Any, Optional
from datetime import datetime, timezone
from typing import Any, Dict, Optional
GC_RETENTION_HARD_MAX = 365
GC_RETENTION_FALLBACK_DEFAULT = 7
@ -109,3 +110,40 @@ def get_gc_retention_days() -> int:
if seed is not None and str(seed).strip():
return clamp_retention_days(seed)
return clamp_retention_days(_default_gc_days())
def task_result_timestamp(result: Dict[str, Any], fallback: Optional[float] = None) -> Optional[float]:
"""The effective task-drive retention date, else the caller's layout-local fallback.
Precedence is the first PARSEABLE value, not the earliest date. A naive date
means UTC. Directory mtime is deliberately not persisted in a task projection:
one task may occupy both layouts, with different directory mtimes.
"""
for key in ("artifact_finalized_at", "completed_at", "finished_at", "ts"):
raw = str(result.get(key) or "").strip()
if not raw:
continue
try:
parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return float(parsed.timestamp())
except ValueError:
continue
return fallback
def task_result_retention_at(result: Dict[str, Any]) -> Optional[str]:
"""A task body's cheap drive-retention projection.
NULL means unclassified/unbindable, '' means no parseable body date, and
other text is normalized epoch seconds. The caller compares it numerically,
never lexically. It selects work; it never authorizes removal.
"""
from ouroboros.task_result_schema import task_result_schema_refusal
status = result.get("status")
if task_result_schema_refusal(result) or not isinstance(status, str) or not status.strip():
return None
timestamp = task_result_timestamp(result)
return "" if timestamp is None else repr(timestamp)

5
ouroboros/settings_scales.py
View file

@ -248,9 +248,8 @@ RESTART_REQUIRED_SETTINGS = frozenset({
"OUROBOROS_HOST_SERVICE_PORT",
# The main interface is selected once per server generation (server_web.ServedUI.apply).
"OUROBOROS_UI_DIR",
# Pooled workers load the extension registry once at spawn and never
# reload it per task; the save-time server reload keeps the skills UI
# fresh, but agent tasks see the new repo only after a restart.
# Workers adopt the server's extension generation before task-start settings
# refresh; a changed repository path still requires a restart of the pool.
"OUROBOROS_SKILLS_REPO_PATH",
"LOCAL_MODEL_SOURCE",
"LOCAL_MODEL_FILENAME",

1
ouroboros/size_ratchet_manifest.py
View file

@ -125,6 +125,7 @@ BAND_PATHS = {
"ouroboros/gateway/extensions.py": "Extensions HTTP surface re-entered the band when the module endpoint moved to the in-memory reviewed bundle (widgets lifecycle 1a); shrink next touch.",
"ouroboros/gateway/host_service.py": "The one loopback callback boundary for reviewed skills: token auth, the chat/decision/presence/WS-relay routes and, with #667, the operation read/cancel that joins existing chat, routing, turn and task records; one trust boundary, one module.",
"ouroboros/gateway/settings.py": "Shrank INTO the band from 1532 lines: the scope-slot window-floor owner capability-ack path was removed with the scope packet (scope-review-retrieval sprint); shrink-only residue, no new content.",
"ouroboros/headless.py": "The bounded drive selector belongs beside the two existing layout pruners, sharing their age, custody, cursor and report contract; extracting a one-caller fragment would split that authority.",
"ouroboros/launcher_bootstrap.py": "Native seed version resync keeps manifest parsing and equal-version payload diagnostics with the existing bootstrap owner; no separate loader or overwrite policy.",
"ouroboros/llm_claudexor.py": "One model-operation owner keeps request, dispatch, result custody and content-free activity observations together; no second polling or state owner.",
"ouroboros/loop_acceptance.py": "the local preparation incident rides the acceptance decision: incident ride-along, the exposed-stance finish intent and the forced-rail cause split live beside the decision authority they qualify (#1223-#1224)",

9
ouroboros/store.py
View file

@ -28,7 +28,7 @@ LOCK_NAME = "current.lock" # the name-tier lock around every access
IMPORT_LOCK_NAME = "current_import.lock"
GENERATION = 8 # the lifecycle generation that owns the store
GENERATION_KEY = "current_store_generation" # in state/migrations.json
SCHEMA_VERSION = 4 # 2: owner questions; 3: acceptance collection; 4: plan collection
SCHEMA_VERSION = 5 # 2: owner questions; 3: acceptance; 4: plans; 5: drive retention
STORE_WAIT_SEC = 10.0
TIER_ENFORCED, TIER_NAME = "enforced", "name"
PUBLISHED, EXPORTING, RETIRED = "published", "exporting", "retired"
@ -63,7 +63,7 @@ CREATE TABLE tasks (
has_pending_promotion INTEGER NOT NULL DEFAULT 0, has_pause_notices INTEGER NOT NULL DEFAULT 0,
delegated_runs_unreconciled INTEGER NOT NULL DEFAULT 0, pending_drive INTEGER NOT NULL DEFAULT 0,
has_owner_questions INTEGER,
review_collection_pending INTEGER, plan_collection_pending INTEGER);
review_collection_pending INTEGER, plan_collection_pending INTEGER, drive_retention_at TEXT);
CREATE INDEX tasks_parent ON tasks(parent_task_id); CREATE INDEX tasks_root ON tasks(root_task_id);
CREATE INDEX tasks_status ON tasks(status); CREATE INDEX tasks_project ON tasks(project_id);
CREATE INDEX tasks_ts ON tasks(ts, task_id); CREATE INDEX tasks_updated ON tasks(updated_at, task_id);
@ -752,11 +752,11 @@ def encode_document(document: Dict[str, Any]) -> str:
def typed_columns(document: Dict[str, Any], *, pending_drive: Optional[bool] = None) -> Dict[str, Any]:
"""The ``tasks`` columns of a document: top-level typed facts, the two Presence facts
derived from their carriers, and the obligation flags from the existing predicates."""
"""Top-level facts and derived Presence, obligation, collection and retention projections."""
from ouroboros.dialogue_provenance import presence_metadata_binding, presence_record_binding
from ouroboros.obligations import result_memberships, review_collection_pending
from ouroboros.plan_record import PLAN_RECORD_KEY, collection_pending
from ouroboros.retention import task_result_retention_at
names = result_memberships(document)
columns: Dict[str, Any] = {}
@ -777,6 +777,7 @@ def typed_columns(document: Dict[str, Any], *, pending_drive: Optional[bool] = N
columns["pending_drive"] = int("nonterminal" in names if pending_drive is None else pending_drive)
columns["review_collection_pending"] = int(review_collection_pending(document))
columns["plan_collection_pending"] = int(collection_pending(document.get(PLAN_RECORD_KEY)))
columns["drive_retention_at"] = task_result_retention_at(document)
return columns

4
ouroboros/store_import.py
View file

@ -138,7 +138,7 @@ def _ensure(root: pathlib.Path, path: pathlib.Path, *, rederive: bool = False) -
def _upgrade_review_collection(txn: Txn) -> None:
"""Schema only: old bodies stay unknown until their ordinary collector visits them.
"""Schema only: old bodies stay unknown until their ordinary maintenance owner visits them.
DDL and its version commit together; an interrupted upgrade is retried in place."""
columns = {row["name"] for row in txn.conn.execute("PRAGMA table_info(tasks)")}
for column, index in (("review_collection_pending", store.REVIEW_COLLECTION_INDEX),
@ -146,6 +146,8 @@ def _upgrade_review_collection(txn: Txn) -> None:
if column not in columns:
txn.conn.execute(f"ALTER TABLE tasks ADD COLUMN {column} INTEGER")
txn.conn.execute(index)
if "drive_retention_at" not in columns:
txn.conn.execute("ALTER TABLE tasks ADD COLUMN drive_retention_at TEXT")
if int(txn.meta("schema_version") or 1) < store.SCHEMA_VERSION:
txn.set_meta("schema_version", store.SCHEMA_VERSION)

59
ouroboros/task_store.py
View file

@ -246,6 +246,65 @@ def task_ids(root: Any) -> List[str]:
_COLLECTION_BATCH = 64 # bounds selection/materialization and one classification commit
DRIVE_RETENTION_BATCH = 64 # bounds directory-name selection and one classification commit
def drive_retention_facts(root: Any, task_ids: Iterable[str], *,
stop: Optional[Callable[[], bool]] = None) -> Dict[str, Dict[str, Any]]:
"""Cheap facts for one bounded batch of drive names, never a scan of task history.
A missing row/body or NULL projection stays a strict-read candidate. Legacy
classification reads coherent bodies outside the selector transaction and
commits only still-NULL write_seq matches, without changing bodies/signatures.
An old body's REPLACE clears the nullable projection even after schema upgrade.
"""
from ouroboros.retention import task_result_retention_at
from ouroboros.task_results import load_task_result
ids = [str(task_id) for task_id in task_ids]
if not ids:
return {}
stopped = stop or (lambda: False)
with open_store(root, write=False) as txn:
schema = txn.meta("schema_version")
if schema is not None and int(schema) < store.SCHEMA_VERSION:
from ouroboros.store_import import upgrade_current_schema
upgrade_current_schema(root)
with open_store(root, write=False) as txn:
rows = txn.conn.execute(
"SELECT t.task_id, t.status, t.delegation_role, t.drive_retention_at, "
"b.task_id IS NOT NULL AS body_present FROM tasks AS t "
"LEFT JOIN task_bodies AS b ON b.task_id=t.task_id "
f"WHERE t.task_id IN ({','.join('?' * len(ids))})", ids).fetchall()
facts = {row["task_id"]: dict(row) for row in rows}
classified = []
for task_id, row in facts.items():
if stopped():
break
if row["drive_retention_at"] is not None or not row["body_present"]:
continue
try:
document, stamp = load_task_result(root, task_id, strict=True, signed=True)
if document is None or stamp is None:
continue
value = task_result_retention_at(document)
if value is None:
continue
# This coherent body's facts improve this pass's hint too. A racing
# writer may still postpone a skip; actual candidates are read afresh.
row.update(status=document["status"], delegation_role=document.get("delegation_role"),
drive_retention_at=value)
classified.append((value, task_id, stamp[0]))
except (store.StoreError, ValueError, TypeError, AttributeError):
# Leave NULL: the pruner's strict read retains the original diagnostic.
continue
if classified and not stopped():
with open_store(root, write=True) as txn:
txn.conn.executemany(
"UPDATE tasks SET drive_retention_at=? WHERE task_id=? AND write_seq=? "
"AND drive_retention_at IS NULL", classified)
return facts
def _collection_ids(root: Any, column: str, pending: Optional[int], after: str) -> List[str]:

302
tests/test_drive_retention_projection.py Normal file
View file

@ -0,0 +1,302 @@
"""Drive pruning selects small current facts before reading possible candidates."""
from __future__ import annotations
import json
import os
import time
import contextlib
from datetime import datetime, timezone
import pytest
from ouroboros import headless, retention, store, store_import, task_store
from ouroboros.task_results import write_task_result
from tests.test_review_collection_index import _legacy_store, _old_replace
from tests.test_store import document, publish
from tests.test_store_import import write_result
NOW = datetime(2026, 10, 16, tzinfo=timezone.utc).timestamp()
OLD = "2026-10-01T00:00:00+00:00"
YOUNG = "2026-10-16T00:00:00+00:00"
FUTURE = "2026-11-01T00:00:00+00:00"
def _layout(root, kind, task_id):
if kind == "headless":
drive = headless.prepare_task_drive(root, task_id, "empty")
return drive.parent, drive, headless.prune_headless_task_drives
drive = root / "task_drives" / task_id
drive.mkdir(parents=True)
return drive, drive, headless.prune_task_drives
def _flag(root, task_id):
with store.open_store(root, write=False) as txn:
return txn.row(task_id, "drive_retention_at", "write_seq")
def _body_spy(monkeypatch):
real, reads = store.Txn.body_text, []
def body(txn, task_id):
raw = real(txn, task_id)
reads.append((task_id, 0 if raw is None else len(raw)))
return raw
monkeypatch.setattr(store.Txn, "body_text", body)
return reads
@pytest.mark.parametrize("kind", ["headless", "direct"])
@pytest.mark.parametrize("count,size", [(3, 128), (70, 100_000)])
def test_real_writer_young_and_nonterminal_bodies_are_not_read_on_either_pass(
tmp_path, monkeypatch, kind, count, size):
root = tmp_path / "data"
for i in range(count):
task_id = f"task-{i:03}"
_directory, drive, prune = _layout(root, kind, task_id)
write_task_result(root, task_id, "running" if i % 2 else "completed",
ts=YOUNG, result="x" * size, child_drive_root=str(drive))
with store.open_store(root, write=False) as txn:
before = txn.revision()
reads = _body_spy(monkeypatch)
for _pass in range(2):
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert report["scanned"] == count
assert not report["pruned"] and not report["errors"]
assert len(report["skipped"]) == count
assert reads == [], "cost must not grow with skipped body size or count"
with store.open_store(root, write=False) as txn:
assert txn.revision() == before, "known hints add no periodic writes"
@pytest.mark.parametrize("kind", ["headless", "direct"])
@pytest.mark.parametrize("fields,removed", [
({"artifact_finalized_at": FUTURE, "completed_at": OLD, "ts": OLD}, False),
({"artifact_finalized_at": "bad", "completed_at": OLD, "ts": FUTURE}, True),
({"artifact_finalized_at": "", "completed_at": "bad",
"finished_at": "2026-10-01T00:00:00", "ts": FUTURE}, True),
({"artifact_finalized_at": OLD, "completed_at": FUTURE, "ts": FUTURE}, True),
({"artifact_finalized_at": "bad", "completed_at": "bad",
"finished_at": "bad", "ts": "bad"}, True),
({"ts": YOUNG}, False),
])
def test_exact_precedence_naive_utc_invalid_dates_and_layout_mtime(
tmp_path, kind, fields, removed):
root = tmp_path / "data"
directory, drive, prune = _layout(root, kind, "task")
publish(root, document("task", child_drive_root=str(drive), **fields))
old = NOW - 20 * 86400
os.utime(directory, (old, old))
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert not report["errors"], report
assert bool(report["pruned"]) is removed
assert directory.exists() is not removed
@pytest.mark.parametrize("kind", ["headless", "direct"])
def test_zero_days_and_cancelled_subagent_exemption_remain_real_settlements(tmp_path, kind):
root = tmp_path / "data"
rows = []
for task_id, status, role, date in (
("before-now", "completed", "root", OLD),
("cancelled-child", "cancelled", "subagent", FUTURE),
("cancelled-root", "cancelled", "root", FUTURE),
):
_directory, drive, prune = _layout(root, kind, task_id)
rows.append(document(task_id, status, delegation_role=role,
child_drive_root=str(drive), ts=date))
publish(root, *rows)
report = prune(root, retention_days=0, now=NOW, live=lambda _: False)
assert {row["task_id"] for row in report["pruned"]} == {"before-now", "cancelled-child"}
assert report["skipped"] == [{"task_id": "cancelled-root", "reason": "younger_than_retention"}]
@pytest.mark.parametrize("kind", ["headless", "direct"])
def test_hint_race_is_rechecked_against_current_body_before_settlement(tmp_path, monkeypatch, kind):
root = tmp_path / "data"
directory, drive, prune = _layout(root, kind, "task")
write_task_result(root, "task", "completed", ts=OLD, child_drive_root=str(drive))
real = task_store.drive_retention_facts
def select_then_enrich(*args, **kwargs):
facts = real(*args, **kwargs)
write_task_result(root, "task", "completed", artifact_finalized_at=FUTURE)
return facts
monkeypatch.setattr(task_store, "drive_retention_facts", select_then_enrich)
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert directory.is_dir() and not report["pruned"]
assert report["skipped"] == [{"task_id": "task", "reason": "younger_than_retention"}]
@pytest.mark.parametrize("tier", [store.TIER_ENFORCED, store.TIER_NAME])
def test_schema_upgrade_is_lazy_and_old_writer_replace_invalidates_hint(tmp_path, monkeypatch, tier):
root = tmp_path / "data"
directory, drive, prune = _layout(root, "direct", "task")
row = document("task", ts=YOUNG, child_drive_root=str(drive), result="old answer" * 1000)
_legacy_store(root, row, owned_root=tmp_path, tier=tier, shape="v3")
reads = _body_spy(monkeypatch)
store_import.upgrade_current_schema(root)
assert _flag(root, "task") == (None, 1)
assert reads == [], "new projection must not backfill bodies during DDL"
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert not report["pruned"] and len(reads) == 1
assert _flag(root, "task") == (retention.task_result_retention_at(row), 1)
reads.clear()
prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert reads == []
# A real predecessor REPLACE uses only its own columns. Meta remains current,
# yet the replacement must lose the new scalar instead of retaining old age.
newer = {**row, "artifact_finalized_at": OLD}
with store.open_store(root, write=True) as txn:
_old_replace(txn.conn, newer, known=(
"has_owner_questions", "review_collection_pending", "plan_collection_pending"))
assert txn.meta("schema_version") == store.SCHEMA_VERSION
assert _flag(root, "task") == (None, 2)
store_import.upgrade_current_schema(root)
assert _flag(root, "task") == (None, 2)
assert prune(root, retention_days=7, now=NOW, live=lambda _: False)["pruned"]
assert not directory.exists()
def test_legacy_classification_is_off_lock_and_cas_keeps_a_racing_writer(tmp_path, monkeypatch):
root = tmp_path / "data"
directory, drive, prune = _layout(root, "direct", "task")
row = document("task", ts=OLD, child_drive_root=str(drive))
_legacy_store(root, row, owned_root=tmp_path, shape="v3")
store_import.upgrade_current_schema(root)
original, touched = retention.task_result_retention_at, []
def classify(document):
if not touched:
touched.append(True)
# Another connection's actual write cannot commit under a held read lock.
with monkeypatch.context() as patch:
patch.setattr(retention, "task_result_retention_at", original)
with store.open_store(root, write=True, wait_sec=0) as txn:
txn.store_task({**document, "artifact_finalized_at": FUTURE})
return original(document)
monkeypatch.setattr(retention, "task_result_retention_at", classify)
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert touched and directory.is_dir() and not report["pruned"]
assert _flag(root, "task") == (
original({**row, "artifact_finalized_at": FUTURE}), 2)
def test_imported_projection_and_quarantined_missing_invalid_outcomes(tmp_path):
root = tmp_path / "data"
for task_id in ("admitted", "missing", "legacy", "invalid"):
_directory, drive, prune = _layout(root, "direct", task_id)
if task_id == "admitted":
write_result(root, document(task_id, ts=YOUNG, child_drive_root=str(drive)))
elif task_id == "legacy":
path = root / "task_results" / f"{task_id}.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({"task_id": task_id, "status": "completed"}), encoding="utf-8")
store_import.ensure_current_store(root, with_children=False)
task_store.patch_document(root, "invalid", lambda _: document("invalid", status="", ts=YOUNG))
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert not report["pruned"]
assert {row["task_id"] for row in report["errors"]} == {"legacy", "invalid"}
assert {row["task_id"]: row["reason"] for row in report["skipped"]} == {
"admitted": "younger_than_retention", "missing": "task_not_terminal"}
assert _flag(root, "admitted")[0] == retention.task_result_retention_at(document("admitted", ts=YOUNG))
@pytest.mark.parametrize("kind", ["headless", "direct"])
def test_unavailable_selector_falls_back_to_strict_real_pruner(tmp_path, monkeypatch, kind):
root = tmp_path / "data"
directory, drive, prune = _layout(root, kind, "task")
write_task_result(root, "task", "completed", ts=OLD, child_drive_root=str(drive))
def unavailable(*_args, **_kwargs):
raise store.StoreBusy("selector unavailable")
monkeypatch.setattr(task_store, "drive_retention_facts", unavailable)
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert report["pruned"] and not report["errors"] and not directory.exists()
@pytest.mark.parametrize("kind", ["headless", "direct"])
def test_budget_counts_settlements_only_cursor_wraps_and_stop_defers(tmp_path, kind):
root = tmp_path / "data"
for task_id, date in (("a", OLD), ("b", YOUNG), ("c", OLD)):
_directory, drive, prune = _layout(root, kind, task_id)
write_task_result(root, task_id, "completed", ts=date, child_drive_root=str(drive))
first = prune(root, retention_days=7, now=NOW, live=lambda _: False, budget=1, after="b")
assert [row["task_id"] for row in first["pruned"]] == ["c"]
assert first["cursor"] == "c" and first["deferred"] == ["a"]
assert first["skipped"] == [{"task_id": "b", "reason": "younger_than_retention"}]
closed = prune(root, retention_days=7, now=NOW, live=lambda _: False, stop=lambda: True, after="c")
assert not closed["pruned"] and closed["deferred"] == ["a"]
assert closed["cursor"] == "c"
last = prune(root, retention_days=7, now=NOW, live=lambda _: False, budget=1, after="c")
assert [row["task_id"] for row in last["pruned"]] == ["a"]
def test_stopped_legacy_classification_leaves_null_without_body_reads(tmp_path, monkeypatch):
root = tmp_path / "data"
_directory, drive, prune = _layout(root, "direct", "task")
row = document("task", ts=OLD, child_drive_root=str(drive))
_legacy_store(root, row, owned_root=tmp_path, shape="v3")
store_import.upgrade_current_schema(root)
reads = _body_spy(monkeypatch)
report = prune(root, retention_days=7, now=NOW, live=lambda _: False, stop=lambda: True)
assert report["deferred"] == ["task"] and not report["pruned"]
assert reads == [] and _flag(root, "task") == (None, 1)
@pytest.mark.parametrize("damage", ["missing_body", "invalid_json", "future_schema"])
def test_unclassified_or_missing_bodies_reach_existing_strict_reader(tmp_path, monkeypatch, damage):
root = tmp_path / "data"
directory, drive, prune = _layout(root, "direct", "task")
write_task_result(root, "task", "completed", ts=YOUNG, child_drive_root=str(drive))
with store.open_store(root, write=True) as txn:
if damage == "missing_body":
txn.conn.execute("DELETE FROM task_bodies WHERE task_id='task'")
else:
raw = "{bad" if damage == "invalid_json" else json.dumps(
{**document("task", ts=YOUNG), "_schema_version": 999})
txn.conn.execute("UPDATE task_bodies SET body=? WHERE task_id='task'", (raw,))
txn.conn.execute("UPDATE tasks SET drive_retention_at=NULL WHERE task_id='task'")
reads = _body_spy(monkeypatch)
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
assert directory.exists() and not report["pruned"]
assert reads and (report["errors"] or report["skipped"])
if damage == "future_schema":
# Store admission, not this selector, owns schema policy. A raw edit of
# an already-admitted row is not re-quarantined by the existing reader.
assert _flag(root, "task")[0] is None
assert report["skipped"] == [{"task_id": "task", "reason": "younger_than_retention"}]
else:
assert not any(row["reason"] == "younger_than_retention" for row in report["skipped"])
@pytest.mark.parametrize("count,size", [(64, 16_384), (512, 65_536)])
def test_real_selection_cost_is_batched_and_independent_of_skipped_body_bytes(tmp_path, monkeypatch, count, size):
root = tmp_path / "data"
for n in range(count):
_directory, drive, prune = _layout(root, "direct", f"task-{n:04}")
write_task_result(root, f"task-{n:04}", "completed", ts=YOUNG,
result="x" * size, child_drive_root=str(drive))
real_open, transactions = task_store.open_store, []
@contextlib.contextmanager
def observed(*args, **kwargs):
transactions.append(kwargs.get("write", False))
with real_open(*args, **kwargs) as txn:
yield txn
reads = _body_spy(monkeypatch)
monkeypatch.setattr(task_store, "open_store", observed)
started, cpu = time.perf_counter(), time.process_time()
report = prune(root, retention_days=7, now=NOW, live=lambda _: False)
elapsed, cpu = time.perf_counter() - started, time.process_time() - cpu
assert len(report["skipped"]) == count and not report["errors"]
assert not reads and not any(transactions)
assert len(transactions) == 2 * ((count + 63) // 64)
print(json.dumps({"tasks": count, "payload_bytes_each": size,
"body_reads": len(reads), "body_chars_read": sum(row[1] for row in reads),
"selector_transactions": len(transactions), "elapsed": elapsed, "cpu": cpu}))

8
tests/test_evidence_pack_fixes.py
View file

@ -426,13 +426,14 @@ def test_stop_latches_and_signals_without_waiting_for_candidate_walk(root, monke
unit._process = child
unit.tick(root, REPO)
def walk(_):
def walk(_, _stop):
entered.set()
assert release.wait(5)
return ["candidate"]
monkeypatch.setattr(packer, "changed_tasks", walk)
monkeypatch.setattr(process_custody, "spawn_supervised", lambda *a, **k: pytest.fail("spawned after Stop"))
monkeypatch.setattr(packer, "packing_pending", walk)
spawns = []
monkeypatch.setattr(process_custody, "spawn_supervised", lambda *a, **k: spawns.append(a) or _FakeChild(b""))
tick = threading.Thread(target=unit.tick, args=(root, REPO))
stop = threading.Thread(target=lambda: (unit.stop(), stopped.set()))
tick.start()
@ -448,6 +449,7 @@ def test_stop_latches_and_signals_without_waiting_for_candidate_walk(root, monke
if unit._thread is not None:
unit._thread.join(3)
assert not tick.is_alive() and not stop.is_alive()
assert spawns == [], "spawned after Stop"
def test_restore_and_preflight_cover_all_descendants_but_not_symlink_drives(root):

8
tests/test_evidence_packer_scaling.py
View file

@ -45,7 +45,9 @@ def test_unrelated_changes_do_not_wake_swept_task(root, monkeypatch, change):
monkeypatch.setattr(packer, "spawn_maintenance_job", lambda *a: pytest.fail("unrelated change spawned a packer"))
assert packer.task_stamps(root, "packed") == before
assert packer.changed_tasks(root) == []
assert not unit.tick(root, REPO)
assert unit.tick(root, REPO)
unit._thread.join(10)
assert not unit._thread.is_alive()
@pytest.mark.parametrize("blocker", ["money", "delivery"])
@ -97,7 +99,9 @@ def test_money_and_deliveries_read_once_for_all_tasks(root, monkeypatch, operati
monkeypatch.setattr(usage_store, "_connect", connect)
monkeypatch.setattr(utils, "read_json_dict", read)
if operation == "tick":
assert not unit.tick(root, REPO)
assert unit.tick(root, REPO)
unit._thread.join(10)
assert not unit._thread.is_alive()
else:
getattr(packer, operation)(root)
queries = [sql for sql in statements if "FROM attempts" in sql and sql.startswith("SELECT")]

4
tests/test_evidence_packs.py
View file

@ -564,7 +564,9 @@ def test_packer_tick_rests_without_candidates_and_the_sweep_remembers_evaluated_
monkeypatch.setattr(process_custody, "spawn_supervised", lambda *a, **k: pytest.fail("spawned without candidates"))
unit = packer.EvidencePacker()
assert unit.tick(root, REPO) is False and unit.tick(root, REPO) is False
assert unit.tick(root, REPO) is False and unit.tick(root, REPO) is True
unit._thread.join(10)
assert not unit._thread.is_alive()
_task(root, "live", blobs=1, status="running")
assert packer.changed_tasks(root) == ["live"]
_run(root) # evaluated, not settled: recorded so the next tick does not re-read it

256
tests/test_maintenance_discovery.py Normal file
View file

@ -0,0 +1,256 @@
"""Maintenance discovery returns the sweep while preserving packer lifetime and packing authority."""
from __future__ import annotations
import json
import sqlite3
import threading
import time
import pytest
from ouroboros import evidence_packer as packer, evidence_packs as packs, store, store_switch
from tests.test_evidence_packs import REPO, _FakeChild, _task, root as root
pytestmark = pytest.mark.serial
def _join(unit):
unit._thread.join(10)
assert not unit._thread.is_alive()
def test_reconcile_releases_its_latch_during_real_stamp_discovery(root, monkeypatch):
from ouroboros import server_maintenance as maintenance
_task(root, "candidate", blobs=1)
unit = packer.EvidencePacker()
assert not unit.tick(root, REPO)
entered, release = threading.Event(), threading.Event()
original = packer._stamp
calls = []
def stamp(*args, **kwargs):
entered.set()
assert release.wait(10)
return original(*args, **kwargs)
monkeypatch.setattr(packer, "_stamp", stamp)
monkeypatch.setattr(packer, "packer", unit)
monkeypatch.setattr(maintenance, "DATA_DIR", root)
monkeypatch.setattr(maintenance, "_stop_requested", lambda *_: False)
monkeypatch.setattr(maintenance, "_periodic_zombie_reconcile", lambda **_: None)
monkeypatch.setattr(maintenance, "_run_drive_custody_pass", lambda *_: None)
monkeypatch.setattr(maintenance, "_step_recovered", calls.append)
for name in ("_STARTUP_PRUNES_OWED", "_STARTUP_TREES_OWED", "_STARTUP_SOURCE_PRUNES_OWED", "_STARTUP_TEMP_SWEEP_OWED"):
monkeypatch.setattr(maintenance, name, [False])
monkeypatch.setattr("supervisor.workspace_preparation.recover_interrupted_preparations", lambda *a, **k: None)
monkeypatch.setattr("ouroboros.late_review_marks.replay_owed_late_reviews", lambda *a, **k: None)
monkeypatch.setattr("ouroboros.review_operation.collect_orphaned_operations_softly", lambda *a, **k: None)
spawns = []
report = json.dumps({"status": "completed", **dict.fromkeys(packer._COUNTERS, 0)}).encode()
monkeypatch.setattr(packer, "spawn_maintenance_job", lambda *a: spawns.append(a) or _FakeChild(report))
latch, marker, finished = threading.Lock(), [0.0], threading.Event()
latch.acquire()
sweep = threading.Thread(target=lambda: (maintenance._run_periodic_reconcile_sweep(marker, latch=latch), finished.set()))
sweep.start()
try:
assert entered.wait(5)
assert finished.wait(1), "candidate discovery still occupies the reconcile pass"
assert not latch.locked() and marker[0] > 0 and calls == ["reconcile_sweep"]
assert not unit.tick(root, REPO), "overlapping discovery admitted"
assert unit.live_children() == [unit._thread]
unit.stop()
assert unit.live_children() == [unit._thread], "Stop is not evidence of reader exit"
finally:
release.set()
sweep.join(10)
_join(unit)
assert unit.live_children() == []
assert spawns == [], "discovery spawned after Stop"
@pytest.mark.parametrize("count", [1, 128])
def test_changed_discovery_short_circuits_but_quiet_install_still_scans(root, monkeypatch, count):
for n in range(count):
_task(root, f"task-{n:04}", blobs=1)
# Publish a valid cache before measuring; no packing child required to read it.
with packs.open_index(root):
pass
calls, original = [], packer._stamp
def stamp(*args, **kwargs):
calls.append(args[1])
return original(*args, **kwargs)
monkeypatch.setattr(packer, "_stamp", stamp)
started, cpu = time.perf_counter(), time.process_time()
assert packer.packing_pending(root, lambda: False)
assert len(calls) == 1
first = {"elapsed": time.perf_counter() - started, "cpu": time.process_time() - cpu}
current = packer.changed_stamps(root)
for task, value in current.items():
packer.record_swept(root, task, value)
calls.clear()
started, cpu = time.perf_counter(), time.process_time()
assert not packer.packing_pending(root, lambda: False)
assert len(calls) == count
print(json.dumps({"tasks": count, "changed_stamps": 1, "quiet_stamps": count,
"first": first, "quiet_elapsed": time.perf_counter() - started,
"quiet_cpu": time.process_time() - cpu}))
def test_archive_only_skips_task_discovery_and_spawns_existing_child(root, monkeypatch):
archive = root / packs.USAGE_ARCHIVE_REL
archive.mkdir(parents=True)
(archive / "segment_ep1_0001.jsonl").write_text("{}\n", encoding="utf-8")
monkeypatch.setattr(packer, "_loose_task_ids", lambda *a: pytest.fail("archive-only needs no task census"))
calls = []
report = json.dumps({"status": "completed", **dict.fromkeys(packer._COUNTERS, 0)}).encode()
monkeypatch.setattr(packer, "spawn_maintenance_job", lambda *a: calls.append(a) or _FakeChild(report))
unit = packer.EvidencePacker()
assert not unit.tick(root, REPO)
assert unit.tick(root, REPO)
_join(unit)
assert len(calls) == 1
def test_discovery_does_not_rebuild_a_missing_index(root, monkeypatch):
_task(root, "candidate", blobs=1)
packs.packs_dir(root).mkdir(parents=True, exist_ok=True)
monkeypatch.setattr(packs, "_rebuild_index", lambda *a: pytest.fail("discovery rebuilt cache"))
assert packer.packing_pending(root, lambda: False)
assert not (packs.packs_dir(root) / packs.INDEX_NAME).exists()
@pytest.mark.parametrize("busy", ["facts", "packs", "packs_sql", "packs_sql_cached", "money"])
def test_real_contention_defers_discovery_without_waiting_or_spawning(root, monkeypatch, busy):
from ouroboros import usage_store
_task(root, "candidate", blobs=1)
with packs.open_index(root):
pass
if busy == "packs_sql_cached":
with packs.open_index(root):
pass
assert str(packs.packs_dir(root) / packs.INDEX_NAME) in packs._VALIDATED
if busy == "money":
with usage_store.hold(root, write=True):
pass
spawns = []
report = json.dumps({"status": "completed", **dict.fromkeys(packer._COUNTERS, 0)}).encode()
monkeypatch.setattr(packer, "spawn_maintenance_job", lambda *a: spawns.append(a) or _FakeChild(report))
unit = packer.EvidencePacker()
assert not unit.tick(root, REPO)
if busy in {"facts", "packs_sql", "packs_sql_cached", "money"}:
path = {"facts": root / "state/current.sqlite", "money": root / usage_store.STORE_REL,
"packs_sql": packs.packs_dir(root) / packs.INDEX_NAME}[busy.removesuffix("_cached")]
conn = sqlite3.connect(path, timeout=0)
conn.execute("BEGIN EXCLUSIVE")
try:
assert unit.tick(root, REPO)
_join(unit)
finally:
conn.rollback()
conn.close()
else:
with store.name_lock(root, packs.INDEX_LOCK_NAME, timeout_sec=0):
assert unit.tick(root, REPO)
_join(unit)
unit.stop()
assert not unit.live_children() and spawns == []
def test_stop_during_directory_discovery_does_not_read_stores(root, monkeypatch):
_task(root, "candidate", blobs=1)
stopped = threading.Event()
original = packer._loose_task_ids
def ids(*args):
result = original(*args)
stopped.set()
return result
monkeypatch.setattr(packer, "_loose_task_ids", ids)
monkeypatch.setattr(store, "open_store", lambda *a, **k: pytest.fail("read after Stop"))
assert not packer.packing_pending(root, stopped.is_set)
def test_actual_exit_consumers_keep_blocked_discovery_visible_then_accept_exit(root, monkeypatch):
import server
from ouroboros import body_adoption
from tests.test_body_adoption_consumers import EXITED
from tests.body_candidate_support import make_serving, make_ctx, rich_commit, restart_receipt
from ouroboros import body_candidate, body_switch
serving = make_serving(root.parent / "body")
monkeypatch.setenv("OUROBOROS_RUNTIME_MODE", "pro")
# These are disposable repositories/data, never the installation.
monkeypatch.setenv("OUROBOROS_SUBAGENT_WORKTREE_ROOT", str(root.parent / "copies"))
ctx = make_ctx(serving, root, "maintenance-adoption")
body_candidate.prepare(ctx)
commit = rich_commit(ctx.repo_dir)
body_candidate.record_reviewed_commit(ctx, commit)
body_adoption.authorize(ctx, commit, reason="test maintenance exit")
restart_receipt(root, commit, "test maintenance exit")
assert body_adoption.bind_restart(lambda **_: (True, "ok"), root, "test maintenance exit")()[0]
_task(root, "candidate", blobs=1)
unit = packer.EvidencePacker()
entered, release = threading.Event(), threading.Event()
original = packer._stamp
def stamp(*args, **kwargs):
entered.set()
assert release.wait(10)
return original(*args, **kwargs)
monkeypatch.setattr(packer, "_stamp", stamp)
monkeypatch.setattr(packer, "packer", unit)
spawns = []
report = json.dumps({"status": "completed", **dict.fromkeys(packer._COUNTERS, 0)}).encode()
monkeypatch.setattr(packer, "spawn_maintenance_job", lambda *a: spawns.append(a) or _FakeChild(report))
monkeypatch.setattr(server, "DATA_DIR", root)
monkeypatch.setattr(server, "begin_owned_stop", lambda *_: None)
monkeypatch.setattr(server, "stop_owned_work", lambda *_: {"state": "completed", "unconfirmed": []})
monkeypatch.setattr(server, "_stop_owned_daemon_for_new_pin", lambda: None)
monkeypatch.setattr("supervisor.workers.kill_workers", lambda **_: None)
monkeypatch.setattr("supervisor.workers.last_worker_exit_census", lambda: EXITED)
monkeypatch.setattr("multiprocessing.active_children", lambda: [])
monkeypatch.setattr("ouroboros.extension_companion.panic_kill_all", lambda: None)
monkeypatch.setattr(server, "_restart_requested", threading.Event())
monkeypatch.setattr(server, "_owner_restart_requested", threading.Event())
server._restart_requested.set()
assert not unit.tick(root, REPO)
assert unit.tick(root, REPO)
try:
assert entered.wait(5)
cleanup = server._emergency_process_cleanup()
assert cleanup["live_children"] == [unit._thread]
assert body_adoption.read(root)["phase"] == "authorized"
assert not pathlib_pointer(serving).exists()
store_switch.record_switch_intent(root, target_sha="target", target_ref="target", reason="test", door="ui_rollback")
assert store_switch.switch_at_restart(root, serving, cleanup)["reason"] == "quiescence_unconfirmed"
finally:
release.set()
_join(unit)
assert spawns == [], "discovery spawned after Stop"
handoff = body_adoption.read(root)
handoff["restart_bound"] = True
body_switch.write_handoff(str(body_adoption.helper_dir(root)), handoff)
cleanup = server._emergency_process_cleanup()
assert not cleanup["live_children"]
assert body_adoption.read(root)["phase"] == "armed"
assert pathlib_pointer(serving).exists()
switches = []
monkeypatch.setattr(store_switch, "switch_needs", lambda *a: {"handles": False, "money": False, "facts": False})
monkeypatch.setattr(store_switch, "rev_parse", lambda *a: "target")
monkeypatch.setattr(store_switch, "switching_owner", lambda *_: lambda intent: (switches.append(intent) or True, "switched"))
# A refused switch withdraws its intent; only a new addressed switch requests it again.
store_switch.record_switch_intent(root, target_sha="target", target_ref="target", reason="test retry", door="ui_rollback")
assert store_switch.switch_at_restart(root, serving, cleanup)["status"] == "switched"
assert len(switches) == 1
def pathlib_pointer(serving):
import pathlib
from ouroboros import body_switch
return pathlib.Path(body_switch.pointer_path(str(serving)))

11
tests/test_maintenance_rest_and_arrivals.py
View file

@ -44,6 +44,7 @@ def env(tmp_path, monkeypatch):
root.mkdir()
monkeypatch.setenv("OUROBOROS_DATA_DIR", str(root))
live, calls, reads, reports = set(), [], [], []
child_run = threading.local()
monkeypatch.setattr(queue, "task_has_live_ownership", lambda task_id, **_kw: task_id in live)
monkeypatch.setattr(queue, "DRIVE_ROOT", root)
@ -66,14 +67,18 @@ def env(tmp_path, monkeypatch):
def spawn(cmd, **_kwargs): # the supervised child, in process: its own ``run_pass``
assert cmd[1:3] == ["-m", "ouroboros.evidence_packer"]
reports.append(packer.run_pass(root))
child_run.active = True
try:
reports.append(packer.run_pass(root))
finally:
child_run.active = False
return _FakeChild(json.dumps(reports[-1]).encode())
monkeypatch.setattr(process_custody, "spawn_supervised", spawn)
monkeypatch.setattr(process_custody, "current_custody_session_id", lambda: "session")
packs._READERS.clear()
yield SimpleNamespace(root=root, live=live, calls=calls, reads=reads, reports=reports, gateway=Gateway,
unit=packer.EvidencePacker(), refs={}, handles={})
unit=packer.EvidencePacker(), refs={}, handles={}, child_run=child_run)
packs._READERS.clear()
@ -233,7 +238,7 @@ def test_late_work_is_handled_by_kind_and_the_history_rests_again(env, monkeypat
entered, release = threading.Event(), threading.Event()
def stat(self, *args, **kwargs):
if self == target and threading.current_thread().name == "evidence-packer" and not entered.is_set():
if (self == target and getattr(env.child_run, "active", False) and not entered.is_set()):
entered.set()
assert release.wait(10)
return original(self, *args, **kwargs)

59
tests/test_maintenance_settings_browser.py Normal file
View file

@ -0,0 +1,59 @@
"""The real Settings counter describes its inspected tail, including zero matches."""
from __future__ import annotations
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
pytest_plugins = ("tests.test_ui_smoke_playwright",)
pytestmark = [pytest.mark.serial, pytest.mark.ui_browser]
@pytest.mark.parametrize("engine", ["chromium", "webkit"])
def test_inspected_safety_event_window_zero_and_nonzero(direct_server_with_data, engine):
from playwright.sync_api import expect, sync_playwright
now = datetime.now(timezone.utc)
sample = {"count": 0}
requests = []
with sync_playwright() as pw:
browser = getattr(pw, engine).launch()
try:
page = browser.new_page(viewport={"width": 1100, "height": 850})
def tail(route):
requests.append(route.request.url)
entries = [{"type": "ordinary_event", "ts": now.isoformat()} for _ in range(1997)]
entries.extend({"type": "safety_mode_skip", "ts": now.isoformat()}
for _ in range(sample["count"]))
entries.extend([
{"type": "safety_mode_skip", "ts": (now - timedelta(days=2)).isoformat()},
{"type": "safety_mode_skip", "ts": "not-a-date"},
])
route.fulfill(json={"entries": entries})
page.route("**/api/logs/events?limit=2000", tail)
for count in (0, 1):
sample["count"] = count
page.goto(direct_server_with_data["url"], wait_until="domcontentloaded")
page.wait_for_selector("#page-chat", timeout=30_000)
page.click('[data-nav-page="settings"]')
expect(page.locator("#btn-save-settings")).to_be_enabled(timeout=30_000)
page.click('[data-settings-tab="behavior"]')
counter = page.locator("#s-safety-skip-counter")
expected = ("1 safety check(s) waved through" if count else
"No waved-through safety checks found")
expect(counter).to_have_text(expected + " among inspected recent events dated within 24h.")
counter.scroll_into_view_if_needed()
assert counter.is_visible()
assert all("limit=2000" in url for url in requests)
evidence = os.environ.get("OUROBOROS_UI_EVIDENCE_OUT")
if evidence:
out = Path(evidence)
out.mkdir(parents=True, exist_ok=True)
page.screenshot(path=str(out / f"safety-window-{engine}-{count}.png"))
assert len(requests) >= 2
finally:
browser.close()

3
tests/test_review_collection_index.py
View file

@ -102,6 +102,7 @@ def _legacy_store(root, *rows, owned_root, tier=store.TIER_ENFORCED, shape="v1")
):
if column not in known:
schema = re.sub(rf",\s*{column} INTEGER", "", schema).replace(index, "")
schema = re.sub(r",\s*drive_retention_at TEXT", "", schema)
conn = store.connect(path, tier, create=True)
try:
conn.executescript(schema)
@ -123,7 +124,7 @@ def _legacy_store(root, *rows, owned_root, tier=store.TIER_ENFORCED, shape="v1")
def _old_replace(conn, row, *, known=()):
columns = store.typed_columns(row)
for column in ("has_owner_questions", "review_collection_pending", "plan_collection_pending"):
for column in ("has_owner_questions", "review_collection_pending", "plan_collection_pending", "drive_retention_at"):
if column not in known:
columns.pop(column)
previous = conn.execute("SELECT write_seq FROM tasks WHERE task_id=?", (row["task_id"],)).fetchone()

12
web/modules/settings.js
View file

@ -771,7 +771,7 @@ export function initSettings({ state, setBeforePageLeave, ws } = {}) {
syncRuntimeModeBridgeState();
syncPolicyState(page, s?._meta);
syncPostTaskEvolutionUi();
refreshSafetySkipCounter(); // fire-and-forget; fills the 24h audited-skip note
refreshSafetySkipCounter(); // fire-and-forget; labels matches in the inspected event window
void refreshSearchPreview(); // fire-and-forget; describes the applied search Source/Model
}
@ -1127,8 +1127,7 @@ export function initSettings({ state, setBeforePageLeave, ws } = {}) {
}
async function refreshSafetySkipCounter() {
// 24h count of durable safety_mode_skip audit events, so the owner sees how much
// the reduced coverage actually waved through.
// Count the last-24h matches inside the returned tail, not all of that day.
const el = byId('s-safety-skip-counter');
if (!el) return;
try {
@ -1139,11 +1138,10 @@ export function initSettings({ state, setBeforePageLeave, ws } = {}) {
const t = Date.parse(String(e?.ts || ''));
return Number.isFinite(t) && t >= cutoff;
}).length;
// Honest window note: the count scans the recent events tail (2000), so a
// very busy day can undercount — say "recent", never overclaim exactness.
// The tail may cover less than a day, including when it has no matches.
el.textContent = n > 0
? `${n} safety check(s) waved through in the last 24h (audited; recent events window).`
: 'No safety checks waved through in the last 24h (recent events window).';
? `${n} safety check(s) waved through among inspected recent events dated within 24h.`
: 'No waved-through safety checks found among inspected recent events dated within 24h.';
} catch {
el.textContent = '';
}