From 9f1a2922fb6f0a8b15959051b18966ce17cad52d Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Thu, 6 Aug 2026 05:40:59 +0000 Subject: [PATCH 1/4] #22 chore: adopt curating-context v1.2 (wave A, pair 3) Bump the vendored skills pointer 549a88b -> 3fc7b71 (the merge carrying curating-context v1.2 and the cohort validation gate) and link the skill through the repo's two-hop vendoring convention. Co-Authored-By: Claude Opus 5 (1M context) --- .claude/skills/curating-context | 1 + skills-vendor/gregoryfoster-skills | 2 +- skills/curating-context | 1 + 3 files changed, 3 insertions(+), 1 deletion(-) create mode 120000 .claude/skills/curating-context create mode 120000 skills/curating-context diff --git a/.claude/skills/curating-context b/.claude/skills/curating-context new file mode 120000 index 0000000..c4920de --- /dev/null +++ b/.claude/skills/curating-context @@ -0,0 +1 @@ +../../skills/curating-context \ No newline at end of file diff --git a/skills-vendor/gregoryfoster-skills b/skills-vendor/gregoryfoster-skills index 549a88b..3fc7b71 160000 --- a/skills-vendor/gregoryfoster-skills +++ b/skills-vendor/gregoryfoster-skills @@ -1 +1 @@ -Subproject commit 549a88b5209868e014dd25e4714f2069334e276e +Subproject commit 3fc7b71e729f4863eed1912df5482f14dde3c5f4 diff --git a/skills/curating-context b/skills/curating-context new file mode 120000 index 0000000..252b6e3 --- /dev/null +++ b/skills/curating-context @@ -0,0 +1 @@ +../skills-vendor/gregoryfoster-skills/skills/curating-context \ No newline at end of file From 8ed86e130c6bdbc18af66a5d00cc2c5a43fdaba2 Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Thu, 6 Aug 2026 06:09:00 +0000 Subject: [PATCH 2/4] #22 docs: curate the context surface to the 6,000-token budget MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AGENTS.md was 19,032 tokens (exact, count_tokens) against a 6,000 budget, with `Bus Conventions` alone at 52% of the file, and all four reference docs orphaned — nothing in the policy file linked any of them. Now 5,773 tokens, under budget, with zero orphans and zero dead links. Every relocated block moved verbatim; prove-no-loss.sh accounts for all 226 non-blank lines of the original (106 inline, 120 relocated, 0 lost). Five new reference docs, all under the 10,000 per-doc budget: docs/ARCHITECTURE.md 8,091 layout tree + the bus-contract reasoning docs/DEPLOYMENT.md 4,232 topology, unit lifecycle, env-var reference docs/STORAGE.md 1,842 blob tree + retention docs/STYLE.md 1,034 the logging stack docs/TESTING.md 905 fakeredis divergences, live-broker fixtures A `## Detail Docs` index links those plus COMMANDS, SKILLS, and both contracts. Nothing was deleted — Phase 2 found no FALSE verdicts, so no warrant existed. The context manifest is re-embedded in the same change, per this repo's own rule that editing a description there is not what a search returns. Also installs the context-budget write guard (PostToolUse), the other half of the ratchet: the weekly run recovers ground, the guard stops regrowth between runs. Co-Authored-By: Claude Opus 5 (1M context) --- .claude/hooks/context-budget-guard.sh | 1 + .claude/settings.json | 12 ++ .skills/context-budget | 1 + .skills/context-doc-budget | 1 + .skills/context-metrics.jsonl | 1 + .skills/context-token-ratio | 1 + .socraticodecontextartifacts.json | 39 ++++- AGENTS.md | 223 +++++++++----------------- docs/ARCHITECTURE.md | 74 +++++++++ docs/DEPLOYMENT.md | 81 ++++++++++ docs/SKILLS.md | 8 + docs/STORAGE.md | 22 +++ docs/STYLE.md | 13 ++ docs/TESTING.md | 16 ++ 14 files changed, 336 insertions(+), 157 deletions(-) create mode 120000 .claude/hooks/context-budget-guard.sh create mode 100644 .skills/context-budget create mode 100644 .skills/context-doc-budget create mode 100644 .skills/context-metrics.jsonl create mode 100644 .skills/context-token-ratio create mode 100644 docs/ARCHITECTURE.md create mode 100644 docs/DEPLOYMENT.md create mode 100644 docs/STORAGE.md create mode 100644 docs/STYLE.md create mode 100644 docs/TESTING.md diff --git a/.claude/hooks/context-budget-guard.sh b/.claude/hooks/context-budget-guard.sh new file mode 120000 index 0000000..28ff9fb --- /dev/null +++ b/.claude/hooks/context-budget-guard.sh @@ -0,0 +1 @@ +../../.claude/skills/curating-context/scripts/context-budget-guard.sh \ No newline at end of file diff --git a/.claude/settings.json b/.claude/settings.json index f647824..aa2f60c 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -17,6 +17,18 @@ } ] } + ], + "PostToolUse": [ + { + "matcher": "Edit|Write|MultiEdit", + "hooks": [ + { + "type": "command", + "command": "bash .claude/hooks/context-budget-guard.sh", + "timeout": 10 + } + ] + } ] }, "permissions": { diff --git a/.skills/context-budget b/.skills/context-budget new file mode 100644 index 0000000..a77fd92 --- /dev/null +++ b/.skills/context-budget @@ -0,0 +1 @@ +6000 diff --git a/.skills/context-doc-budget b/.skills/context-doc-budget new file mode 100644 index 0000000..5caff40 --- /dev/null +++ b/.skills/context-doc-budget @@ -0,0 +1 @@ +10000 diff --git a/.skills/context-metrics.jsonl b/.skills/context-metrics.jsonl new file mode 100644 index 0000000..23c7af8 --- /dev/null +++ b/.skills/context-metrics.jsonl @@ -0,0 +1 @@ +{"actions": ["demote:Bus Conventions", "demote:Environment Variables", "demote:Project Layout", "demote:Server Lifecycle", "demote:Infrastructure", "demote:Retention", "demote:Conventions/logging", "index:Detail Docs", "relink:orphans"], "budget": 6000, "bytes": 14254, "delta_days": null, "delta_tokens": null, "docs_orphaned": 0, "docs_total": 9, "file": "AGENTS.md", "lines": 232, "links_dead": 0, "no_loss": "ok", "note": null, "over_budget": false, "repo": "replicator", "skill_commit": "3fc7b71", "skill_version": "1.2", "tokens": 5773, "tokens_exact": true, "tokens_live": 50759, "top_section": "Code Exploration Policy", "top_section_share": 18, "ts": "2026-08-06"} diff --git a/.skills/context-token-ratio b/.skills/context-token-ratio new file mode 100644 index 0000000..e72716a --- /dev/null +++ b/.skills/context-token-ratio @@ -0,0 +1 @@ +2.46 diff --git a/.socraticodecontextartifacts.json b/.socraticodecontextartifacts.json index 24b8f58..e2639a9 100644 --- a/.socraticodecontextartifacts.json +++ b/.socraticodecontextartifacts.json @@ -3,7 +3,7 @@ { "name": "agent-guidelines", "path": "./AGENTS.md", - "description": "Authoritative project guide for agents and contributors. Worker-first shape (the bus consumer is the primary process; FastAPI /health is dev-only), the co-core wheelhouse mechanism (gs://co-gcs-pypi via [tool.uv] find-links, not PyPI), src/ layout, the REPLICATOR_-prefixed env contract and the hard boundary between /etc/replicator/.env and the repo .env, the Redis >=7.0 floor and Archiver's ownership of the broker, bus consumer conventions (at-least-once idempotency, schema_version branching, the read(count>1) batch-poison caveat, the DLQ seam), the fake_redis test fixture, and commit conventions." + "description": "Authoritative project guide for agents and contributors, curated to a 6,000-token budget: it carries the rules needed on nearly every task and routes to the reference docs for everything else (see its Detail Docs index). Inline: the worker-first shape (the bus consumer is the primary process; FastAPI /health is dev-only), the co-core wheelhouse mechanism (gs://co-gcs-pypi via [tool.uv] find-links, not PyPI) and the load-bearing patch floor, the SocratiCode code-exploration policy, the hard boundary between /etc/replicator/.env and the repo .env, the Redis >=7.0 floor and Archiver's ownership of the broker, the nine bus rules that bind every change (at-least-once idempotency with two keys, store-then-publish, read(count=1), fail-loud from_wire with a global dispatch table, DLQ vs retry, no outbox on the consume path, the three stream kinds, seed_fetch.py as the only writer to content.fetch, the two normative contracts), the common commands, and the commit/logging/date conventions. Module layout, the reasoning behind each bus contract, blob-tree and retention semantics, deployment topology and the full environment-variable reference, fakeredis divergences, and the logging stack live in docs/ARCHITECTURE.md, docs/STORAGE.md, docs/DEPLOYMENT.md, docs/TESTING.md, and docs/STYLE.md respectively." }, { "name": "architecture", @@ -13,17 +13,42 @@ { "name": "design-plans", "path": "./docs/plans/", - "description": "Founding MVP design (2026-06-25, adopted). The command/fact contracts as shipped in co-core v0.7.0 \u2014 ContentFetchCommand (URL-addressed, command_id as the wire idempotency key) and BlobAvailableEvent (content_fingerprint, blob_uri, size_bytes, media_type, url, optional command_id) \u2014 the stream taxonomy and DLQ naming, the temp-storage interface, the two-level idempotency argument, why fingerprint parity dissolves when Replicator is the sole fetcher, explicit MVP scope cuts, and the build sequence. The founding plan's three open questions were settled 2026-07-31 in a companion doc: blob_uri is a file:// URI over a two-level sharded, content-addressed path (///.bin, constant .bin extension, temp-file + os.replace so presence means complete); the MVP command issuer is an in-repo scripts/seed_fetch.py rather than Watcher or Archiver; and the MVP boundary is stop-at-fact, with archiver-writeback reclassified from MVP+ to won't-do \u2014 Archiver consumes blob_available and writes the SourceRevision itself (archiver#118, which also carries the info_source_id contract gap). Blob retention is out of MVP scope and tracked in #5." + "description": "Founding MVP design (2026-06-25, adopted). The command/fact contracts as shipped in co-core v0.7.0 — ContentFetchCommand (URL-addressed, command_id as the wire idempotency key) and BlobAvailableEvent (content_fingerprint, blob_uri, size_bytes, media_type, url, optional command_id) — the stream taxonomy and DLQ naming, the temp-storage interface, the two-level idempotency argument, why fingerprint parity dissolves when Replicator is the sole fetcher, explicit MVP scope cuts, and the build sequence. The founding plan's three open questions were settled 2026-07-31 in a companion doc: blob_uri is a file:// URI over a two-level sharded, content-addressed path (///.bin, constant .bin extension, temp-file + os.replace so presence means complete); the MVP command issuer is an in-repo scripts/seed_fetch.py rather than Watcher or Archiver; and the MVP boundary is stop-at-fact, with archiver-writeback reclassified from MVP+ to won't-do — Archiver consumes blob_available and writes the SourceRevision itself (archiver#118, which also carries the info_source_id contract gap). Blob retention is out of MVP scope and tracked in #5." }, { "name": "issuer-contract", "path": "./docs/contracts/content-fetch-issuer-contract.md", - "description": "Normative contract for any service publishing content.fetch commands (seed_fetch.py today, Watcher from Phase 4) \u2014 and its permanent home, linked to rather than copied by issuer-side repos. Opens with the frame shape, because getting it wrong is the likeliest and quietest mistake: what lands on the stream is a co-core envelope produced by to_wire (key, payload as JSON, event_type, schema_version, occurred_at, content_type), never the model's fields flattened, and an XADD of top-level command_id/url fails from_wire into content.fetch.dlq silently. key is derived by to_wire (command_id for a command, content_fingerprint for a fact) and is not what Replicator dedupes on \u2014 that is the decoded payload.command_id \u2014 but it is what makes a DLQ entry correlatable, so an issuer can tail content.fetch.dlq with a plain XREAD and close a pending entry deterministically instead of waiting out a timeout. Then why the wire carries no info_source_id and what that pushes onto the issuer: mint a fresh ULID command_id per fetch occasion (a resource-stable id is silently deduped away for 24h by replicator:cmd:); url is one-to-many onto InfoSources and therefore not a correlation key; persist the command_id -> domain map before publishing, outbox-style; correlation must be idempotent because a crash between publish and the dedupe SET emits a second fact under the same command_id; consumers must not dedupe their inbox on content_fingerprint, which is storage identity rather than correlation identity; there is no fetch_failed fact, so silence is the only failure signal and the issuer needs its own pending-entry reaper that re-issues rather than concluding loss; blob_uri is a host-local file:// URI whose 7-day TTL runs from last fetch-reference, not last consumer read. Carries the full failure taxonomy (which conditions dead-letter to content.fetch.dlq silently, which retry indefinitely, which park in the PEL under the byte ceiling), what Replicator does and does not guarantee, the unauthenticated-capability trust posture resting on Archiver's localhost bus ACL, and the deferred fetch_failed fact as a co-core change. Written for #8." + "description": "Normative contract for any service publishing content.fetch commands (seed_fetch.py today, Watcher from Phase 4) — and its permanent home, linked to rather than copied by issuer-side repos. Opens with the frame shape, because getting it wrong is the likeliest and quietest mistake: what lands on the stream is a co-core envelope produced by to_wire (key, payload as JSON, event_type, schema_version, occurred_at, content_type), never the model's fields flattened, and an XADD of top-level command_id/url fails from_wire into content.fetch.dlq silently. key is derived by to_wire (command_id for a command, content_fingerprint for a fact) and is not what Replicator dedupes on — that is the decoded payload.command_id — but it is what makes a DLQ entry correlatable, so an issuer can tail content.fetch.dlq with a plain XREAD and close a pending entry deterministically instead of waiting out a timeout. Then why the wire carries no info_source_id and what that pushes onto the issuer: mint a fresh ULID command_id per fetch occasion (a resource-stable id is silently deduped away for 24h by replicator:cmd:); url is one-to-many onto InfoSources and therefore not a correlation key; persist the command_id -> domain map before publishing, outbox-style; correlation must be idempotent because a crash between publish and the dedupe SET emits a second fact under the same command_id; consumers must not dedupe their inbox on content_fingerprint, which is storage identity rather than correlation identity; there is no fetch_failed fact, so silence is the only failure signal and the issuer needs its own pending-entry reaper that re-issues rather than concluding loss; blob_uri is a host-local file:// URI whose 7-day TTL runs from last fetch-reference, not last consumer read. Carries the full failure taxonomy (which conditions dead-letter to content.fetch.dlq silently, which retry indefinitely, which park in the PEL under the byte ceiling), what Replicator does and does not guarantee, the unauthenticated-capability trust posture resting on Archiver's localhost bus ACL, and the deferred fetch_failed fact as a co-core change. Written for #8." }, { "name": "boundaries-charter", "path": "./docs/contracts/replicator-boundaries.md", - "description": "Normative charter for what Replicator is allowed to become \u2014 sibling to the issuer contract, which settles the wire while this settles the service. The rule: Replicator owns the mechanics of acquiring bytes from a network and holding them briefly, never why, when, or what they mean. Three tests for any proposed capability, field, or setting (durable per-resource history -> issuer; cross-command coordination over a resource only the fetcher can see -> Replicator; expressible without domain vocabulary or it is the issuer's), resolved when both fire by mechanism-to-Replicator / policy-to-the-issuer / config-over-the-bus. Worked examples: per-host politeness (the hard case, where the tests do not resolve to one side) and conditional GET (ETag/Last-Modified are durable per-resource history, so the fact returns them and the next command replays them as headers \u2014 conditional GET with zero per-URL state here). Config taxonomy of three channels (env for host facts, command for the occasion, a policy stream for cluster policy needing cross-command state) with a fourth \u2014 an inbound admin HTTP API \u2014 rejected by name; ingress is read-only liveness and self-description, and the invariant that matters is asserted about the worker, which binds no port, not the dev-only FastAPI app. The policy stream SHIPPED in #19 on co-core v0.7.7 (cannobserv#285) and is named **content.fetch-policy with a hyphen** \u2014 the dotted content.fetch.policy this document originally proposed collides with the .dlq derivation of the command stream. Last-write-wins per host, replayed from 0-0 at boot via the groupless AsyncBusTailReader and tailed thereafter, MAXLEN-trimmed by the producer because periodic full republication onto an untrimmed stream is unbounded growth, a Redis hash rejected for having no schema_version and coupling to a key name. Documents the consumer-side rules that fail silently if missed: an unknown OR revoked host resolves to the conservative env default and never to unlimited (revoked is a tombstone meaning 'no explicit policy', so branch on revoked before reading min_interval_seconds, which is None there); 0.0 is a legal interval and is falsy; the default's strictness cannot be asserted at startup because a published interval has no upper bound, so it is reported per host at apply time instead; arrival order is not publication order, so the map guards on occurred_at per host; and from_wire's dispatch table is global, so a foreign payload decodes cleanly into the wrong model and needs an isinstance check. Enforcement idiom: park in the PEL, return via claim_stale \u2014 whose 60s granularity misses Watcher's 1.0s baseline by 60x, resolved by splitting the wait by duration (sleep under one poll window, park above it). REPLICATOR_MIN_HOST_INTERVAL_SECONDS is now the FALLBACK rather than the rule, and 0 no longer disables pacing outright. Two known limitations recorded rather than omitted: blob_uri's host-local file:// coupling (#7), and the host asked for not being the host reached under redirects, which #19 did NOT resolve. Enforced by tests/test_boundaries.py in CI: eight invariants including an AST scan of src/ for domain nouns in identifiers and string literals, a recursive route walk, no locally-defined wire models (payload shapes, not producer-owned token vocabularies like FailureReason), and REPLICATOR_-prefixed settings with BUILD_ID exempted by name. Written for #12, updated for #19." + "description": "Normative charter for what Replicator is allowed to become — sibling to the issuer contract, which settles the wire while this settles the service. The rule: Replicator owns the mechanics of acquiring bytes from a network and holding them briefly, never why, when, or what they mean. Three tests for any proposed capability, field, or setting (durable per-resource history -> issuer; cross-command coordination over a resource only the fetcher can see -> Replicator; expressible without domain vocabulary or it is the issuer's), resolved when both fire by mechanism-to-Replicator / policy-to-the-issuer / config-over-the-bus. Worked examples: per-host politeness (the hard case, where the tests do not resolve to one side) and conditional GET (ETag/Last-Modified are durable per-resource history, so the fact returns them and the next command replays them as headers — conditional GET with zero per-URL state here). Config taxonomy of three channels (env for host facts, command for the occasion, a policy stream for cluster policy needing cross-command state) with a fourth — an inbound admin HTTP API — rejected by name; ingress is read-only liveness and self-description, and the invariant that matters is asserted about the worker, which binds no port, not the dev-only FastAPI app. The policy stream SHIPPED in #19 on co-core v0.7.7 (cannobserv#285) and is named **content.fetch-policy with a hyphen** — the dotted content.fetch.policy this document originally proposed collides with the .dlq derivation of the command stream. Last-write-wins per host, replayed from 0-0 at boot via the groupless AsyncBusTailReader and tailed thereafter, MAXLEN-trimmed by the producer because periodic full republication onto an untrimmed stream is unbounded growth, a Redis hash rejected for having no schema_version and coupling to a key name. Documents the consumer-side rules that fail silently if missed: an unknown OR revoked host resolves to the conservative env default and never to unlimited (revoked is a tombstone meaning 'no explicit policy', so branch on revoked before reading min_interval_seconds, which is None there); 0.0 is a legal interval and is falsy; the default's strictness cannot be asserted at startup because a published interval has no upper bound, so it is reported per host at apply time instead; arrival order is not publication order, so the map guards on occurred_at per host; and from_wire's dispatch table is global, so a foreign payload decodes cleanly into the wrong model and needs an isinstance check. Enforcement idiom: park in the PEL, return via claim_stale — whose 60s granularity misses Watcher's 1.0s baseline by 60x, resolved by splitting the wait by duration (sleep under one poll window, park above it). REPLICATOR_MIN_HOST_INTERVAL_SECONDS is now the FALLBACK rather than the rule, and 0 no longer disables pacing outright. Two known limitations recorded rather than omitted: blob_uri's host-local file:// coupling (#7), and the host asked for not being the host reached under redirects, which #19 did NOT resolve. Enforced by tests/test_boundaries.py in CI: eight invariants including an AST scan of src/ for domain nouns in identifiers and string literals, a recursive route walk, no locally-defined wire models (payload shapes, not producer-owned token vocabularies like FailureReason), and REPLICATOR_-prefixed settings with BUILD_ID exempted by name. Written for #12, updated for #19." + }, + { + "name": "architecture-detail", + "path": "./docs/ARCHITECTURE.md", + "description": "Module-by-module layout of src/ (worker, storage, api, core) with the job each file owns, and the full reasoning behind every bus contract Replicator consumes or publishes: the issuer contract and boundaries charter pointers, content.blobs carrying both blob_available and fetch_failed, the six optional passthrough fields on blob_available and why None means nobody said it, the FailureReporter seam, per-command headers/timeout_seconds validation and why unsendable options are refused rather than clamped, the pacing split between sleeping and parking, the groupless content.fetch-policy reader and the four ways to apply a policy message wrongly, at-least-once idempotency, claim_stale as the retry path, XPENDING delivery accounting, and the message-vs-cycle failure distinction. Read this before changing what any stream carries." + }, + { + "name": "temp-storage", + "path": "./docs/STORAGE.md", + "description": "The blob tree: file://///.bin with a constant .bin extension, temp-file + os.replace so presence means complete, and creation-only 0644/0755 modes with the startup reachability warning. Retention: the TTL runs from last reference (os.utime on the store short-circuit), the three populations under REPLICATOR_BLOB_DIR and why in-flight .tmp files are not garbage, the byte ceiling as backpressure rather than a faster clock, the single BlobUsage with two writers, orphan recording, and why the sweep runs in the worker rather than a systemd timer." + }, + { + "name": "deployment-reference", + "path": "./docs/DEPLOYMENT.md", + "description": "Single-VM topology and port allocation (worker binds none, 8040 reserved, 8041 dev; neighbours watcher 8000/8001, archiver 8020/8021, notifier 9000/9001), Archiver's ownership of the Redis broker and the >=7.0 server floor, the replicator.service lifecycle table, why /etc/systemd/system/replicator.service is a copy rather than a symlink, the dev-server invocation, and the complete REPLICATOR_-prefixed environment-variable reference with the reasoning behind every default." + }, + { + "name": "testing-reference", + "path": "./docs/TESTING.md", + "description": "How the bus is tested: the fake_redis fixture and what it is sound for, the two places fakeredis diverges from the live broker (it registers a consumer on an empty XREADGROUP; it ignores block), the real_redis fixture with its db-0 refusal and auth-failure re-raise, the scratch_topic confinement, and the one namespace outside replicator.itest.* that an end-to-end run creates." + }, + { + "name": "style-logging", + "path": "./docs/STYLE.md", + "description": "The logging stack: build_json_formatter() as the single schema definition installed two ways (configure_logging on the root logger, and src/core/log_config.json naming the same factory for uvicorn), why the worker needs no --log-config, the two plain-text journald lines the wheelhouse-sync ExecStartPre writes and why they cannot be JSON, and why the color_message strip is a filter on the loggers rather than on the formatter or the handler." }, { "name": "commands", @@ -43,17 +68,17 @@ { "name": "wheelhouse-sync", "path": "./scripts/sync_wheelhouse.py", - "description": "Mirrors the private cannobserv package index (gs://co-gcs-pypi/wheels/) into ./.wheelhouse, from which uv resolves co-core and co-core-aio via find-links. Must run before uv sync and must not import the project, since the project's deps are what it provides \u2014 hence the isolated `uv run --no-project --with google-cloud-storage` invocation. ADC auth: the SA key on the VM, keyless WIF in CI." + "description": "Mirrors the private cannobserv package index (gs://co-gcs-pypi/wheels/) into ./.wheelhouse, from which uv resolves co-core and co-core-aio via find-links. Must run before uv sync and must not import the project, since the project's deps are what it provides — hence the isolated `uv run --no-project --with google-cloud-storage` invocation. ADC auth: the SA key on the VM, keyless WIF in CI." }, { "name": "redis-floor-guard", "path": "./scripts/check_redis_floor.sh", - "description": "ExecStartPre guard asserting the Redis >=7.0 change-bus server floor. Replicator is the cluster's first user of AsyncBusConsumer.claim_stale, which reads XAUTOCLAIM's three-element reply \u2014 the deleted-ids element added in Redis server 7.0 \u2014 so a lower version breaks crash recovery. Soft on an unreachable broker (it may still be starting), fatal on a genuine downgrade." + "description": "ExecStartPre guard asserting the Redis >=7.0 change-bus server floor. Replicator is the cluster's first user of AsyncBusConsumer.claim_stale, which reads XAUTOCLAIM's three-element reply — the deleted-ids element added in Redis server 7.0 — so a lower version breaks crash recovery. Soft on an unreachable broker (it may still be starting), fatal on a genuine downgrade." }, { "name": "ci-pipeline", "path": "./.github/workflows/ci.yml", - "description": "GitHub Actions CI. Keyless Workload Identity Federation auth to GCP for the wheelhouse mirror, a fail-fast guard asserting the GCP_WIF_PROVIDER org variable is visible to this repo, ruff check and format gates, a co-core extras import smoke proving [extract] and [bus] wire up in a clean environment, and pytest with the coverage gate. Deliberately no Postgres service and no alembic steps \u2014 Replicator is DB-free." + "description": "GitHub Actions CI. Keyless Workload Identity Federation auth to GCP for the wheelhouse mirror, a fail-fast guard asserting the GCP_WIF_PROVIDER org variable is visible to this repo, ruff check and format gates, a co-core extras import smoke proving [extract] and [bus] wire up in a clean environment, and pytest with the coverage gate. Deliberately no Postgres service and no alembic steps — Replicator is DB-free." }, { "name": "dependency-contract", diff --git a/AGENTS.md b/AGENTS.md index bb7e861..992bae4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -61,79 +61,35 @@ Prefetch query — run via `ToolSearch` at session start: ## Project Layout -``` -src/worker/ — Bus consumer; the primary process -src/worker/main.py — Entry point: client lifetime, consumer group, signals -src/worker/loop.py — The consume path: poll → dispatch → ack, DLQ, dedupe, recovery -src/worker/handler.py — The byte path behind the Handler seam: fetch → fingerprint → store → publish -src/worker/reporter.py — The failure fact behind the FailureReporter seam: fetch_failed on content.blobs -src/worker/retention.py — The sweep task: cadence, usage accounting, ceiling reporting -src/worker/pacing.py — Per-host request spacing; the mechanism half of politeness (#12) -src/worker/policy.py — The content.fetch-policy consumer: the map, and the groupless tail (#19) -src/storage/ — Temp storage; BlobStore protocol + the local-FS backend -src/storage/base.py — BlobStore protocol (store / exists / open) -src/storage/local.py — Content-addressed local backend; file:// URIs, sharded paths -src/storage/sweeper.py — Retention: TTL reap, stale temps, empty shards; the measured size -src/api/ — FastAPI app (/health only; not part of the MVP loop) -src/api/main.py — App factory, lifespan, router registration -src/core/ — Shared domain logic, logging, config -src/core/errors.py — TransientFetchError / PermanentFetchError + FailureReason (handler failure vocabulary) -src/core/logging.py — build_json_formatter() + ColorMessageFilter + configure_logging() + get_logger() -src/core/log_config.json — uvicorn --log-config; routes uvicorn's own loggers through that formatter -src/core/config.py — Settings / env access (see Environment Variables) -scripts/ — sync_wheelhouse.py, check_redis_floor.sh, seed_fetch.py -scripts/seed_fetch.py — the MVP command issuer; publishes content.fetch, --watch tails the facts -tests/ — Mirrors src/ structure; integration tests in `@pytest.mark.integration` -docs/ — Reference docs (COMMANDS, SKILLS) -docs/contracts/ — Normative contracts, linked to from sibling repos: the issuer-facing half of the wire, and the boundaries charter (what Replicator may become) -docs/plans/ — Implementation plans -deploy/ — Systemd unit + deployment config -.wheelhouse/ — Local mirror of the private cannobserv index (git-ignored except .gitkeep) -``` +`src/worker/` is the primary process — the bus consumer, with the byte path +(`handler.py`), the failure fact (`reporter.py`), the retention sweep, the pacer, +and the `content.fetch-policy` reader behind their own seams. `src/storage/` is +the content-addressed temp store behind the `BlobStore` protocol; `src/api/` is +the dev-only `/health` app; `src/core/` holds config, logging, and the handler's +failure vocabulary. `tests/` mirrors `src/`. Every module with the job it owns: +[docs/ARCHITECTURE.md](docs/ARCHITECTURE.md). ## Infrastructure **Single-VM setup.** Code committed to main is the deployed code. Replicator shares the VM with archiver, watcher, and notifier. -| Service | Framework | Port | Managed by | -|---|---|---|---| -| Worker (live) | asyncio bus consumer | — | `systemctl` (`replicator.service`) | -| API (dev) | FastAPI | 8041 | manual uvicorn | - -The worker binds no port. Port 8040 is reserved for Replicator's API should it ever be deployed; 8041 is the dev port. Neighbours: watcher 8000/8001, archiver 8020/8021, notifier 9000/9001. The exe.dev proxy transparently forwards ports 3000–9999; the dev server is reachable at `https://replicator.exe.xyz:8041/`. - -### Redis is Archiver-operated — Replicator connects, it does not run its own - -The Redis change bus is Archiver-operated cluster infrastructure (the shared VM's `redis-server.service`). Replicator is a **client**: never ship a broker, never claim ownership. - -**Redis ≥ 7.0 is Replicator-critical.** Replicator is the cluster's first user of `AsyncBusConsumer.claim_stale`, which reads `XAUTOCLAIM`'s three-element reply — the deleted-ids element added in Redis **server** 7.0. Below that, the crash-recovery path raises. `scripts/check_redis_floor.sh` guards this as an `ExecStartPre`. (The VM runs 7.0.15.) - -The **redis-py client** resolves `>=5,<8` transitively via `co-core-aio[bus]`. Don't re-pin it narrower. +The worker binds no port; 8041 is the dev API port and 8040 is reserved. **Redis +is Archiver-operated** — Replicator is a client, never ships a broker, never +claims ownership — and server **≥ 7.0** is Replicator-critical because +`claim_stale` reads `XAUTOCLAIM`'s three-element reply. `scripts/check_redis_floor.sh` +guards it as an `ExecStartPre`. Ports, neighbours, and the redis-py pin: +[docs/DEPLOYMENT.md](docs/DEPLOYMENT.md). ## Server Lifecycle -**`replicator.service` runs the worker.** It binds no port, so there is no port to conflict over — but only one process should hold a given consumer name at a time. - -| Situation | Action | -|---|---| -| Code committed to main | `uv sync --frozen && sudo systemctl restart replicator` | -| Testing a worktree/branch | `uv run python -m src.worker.main` (set a distinct `REPLICATOR_CONSUMER_NAME`) | -| Debugging the live service | `sudo journalctl -u replicator -f` | -| After editing `deploy/replicator.service` | `sudo cp deploy/replicator.service /etc/systemd/system/ && sudo systemctl daemon-reload && sudo systemctl restart replicator` | -| After a co-core version bump | re-run `sync_wheelhouse.py`, then `uv sync` | - -`ExecStart` uses `--frozen --no-sync`, so dependency sync is a deploy step, not a service-start side effect. - -**`/etc/systemd/system/replicator.service` is a *copy*, not a symlink to `deploy/`.** So the `cp` above is load-bearing and `daemon-reload` alone silently does nothing — systemd re-reads the installed file, which is still the old one. The failure has no symptom at restart: the worker comes up on the new code under the *old* unit, and the mismatch only surfaces the first time a directive actually matters. Nothing guards it, either — `tests/test_deploy.py` reads the repo file, which is exactly the copy that is still correct. Diff the two when a restart follows a unit edit (#11 deploy). - -The copy is deliberate, for the same reason `/etc/replicator/.env` is not read from the repo: the live unit must survive a repo reset, a worktree switch, or a branch checkout that happens to be mid-edit. - -**Dev server workflow** (the `/health` app, port 8041 so a future live service stays up): - -```bash -set -a; . /etc/replicator/.env 2>/dev/null; . .env 2>/dev/null; set +a -uv run uvicorn src.api.main:app --host 0.0.0.0 --port 8041 --reload --log-config src/core/log_config.json -``` +`replicator.service` runs the worker. Deploy committed code with `uv sync --frozen +&& sudo systemctl restart replicator`; debug with `sudo journalctl -u replicator -f`; +test a branch with `uv run python -m src.worker.main` under a distinct +`REPLICATOR_CONSUMER_NAME`. **After editing `deploy/replicator.service`, `cp` it to +`/etc/systemd/system/`** — the installed unit is a copy, not a symlink, so +`daemon-reload` alone silently re-reads the old file and the mismatch has no +symptom until a directive matters. Full lifecycle table and the dev-server +invocation: [docs/DEPLOYMENT.md](docs/DEPLOYMENT.md). ## Environment Variables @@ -152,91 +108,49 @@ set -a; . /etc/replicator/.env 2>/dev/null; . .env 2>/dev/null; set +a Replicator-owned settings carry the `REPLICATOR_` prefix so they never collide with a sibling service on the shared VM. `BUILD_ID` is deliberately unprefixed — the systemd unit stamps it generically. -In `.env` (dev/agent only — never read by the service): -- `GH_TOKEN` — GitHub PAT for this repo (used by `gh` CLI) -- `GH_TOKEN_ARCHIVER` / `GH_TOKEN_WATCHER` / `GH_TOKEN_CANNOBSERV` / `GH_TOKEN_SKILLS` — per-repo PATs. Cross-repo work is **filed as an issue**, never edited directly: each repo owns its own review, CI, and deploy cycle, and `main` is the deployed code. Pass the right one as `GH_TOKEN` for a given `gh` call. - -Read by neither env file — test-only, defined in `tests/conftest.py`: -- `REPLICATOR_TEST_REDIS_URL` — live broker for `@pytest.mark.integration`; default `redis://localhost:6379/15`. Must not resolve to db 0 (the fixture fails outright if it does) — see **Testing the bus** - -In `/etc/replicator/.env` (read by the service): -- `GOOGLE_APPLICATION_CREDENTIALS` — SA key for the wheelhouse mirror (`/etc/replicator/co-pypi-reader.json`) -- `REPLICATOR_REDIS_URL` — change-bus client URL; default `redis://localhost:6379/0` -- `REPLICATOR_BLOB_DIR` — temp-storage root for fetched bytes; default `blobs`. Resolved to an absolute path at store construction — `file://` URIs require it -- `REPLICATOR_BLOB_TTL_SECONDS` — how long a blob survives after it was **last referenced**; default `604800` (7 days). Measured from mtime, which the store refreshes on its short-circuit. The number is a published commitment to archiver (archiver#118), not a local tuning knob — raise it if a `content.blobs` consumer says it needs longer -- `REPLICATOR_BLOB_SWEEP_INTERVAL_SECONDS` — how often the tree is walked; default `900`. Also the staleness bound on the measured byte total the ceiling reads -- `REPLICATOR_BLOB_TEMP_GRACE_SECONDS` — how long a `.tmp` may live before the sweep treats it as debris; default `3600`. Deliberately unrelated to the TTL and far shorter — see **Retention** -- `REPLICATOR_BLOB_MAX_TOTAL_BYTES` — ceiling on everything the blob tree holds; default `2147483648` (2 GiB). Crossing it pauses fetching (`TransientFetchError`); it never shortens the TTL -- `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` — the most a command's own `timeout_seconds` may ask for; default `120`. Not a default (an omitted field still gets the driver's 30 s) but a ceiling, and a guard rather than a preference: the consume path is serial, so one issuer's timeout is a lien on every other command in the group. Over it ⇒ `PermanentFetchError`. Bounded above by the unit's `TimeoutStopSec` — change one, revisit the other -- `REPLICATOR_MAX_BLOB_BYTES` — ceiling on one fetched body; default `67108864` (64 MiB). A **storage** guard, not a memory one: co-core's fetch driver buffers the whole response before returning it, so the bytes are already resident when this is checked. Over the ceiling ⇒ `PermanentFetchError` ⇒ DLQ -- `REPLICATOR_CONSUMER_GROUP` — consumer group on `content.fetch`; default `replicator.fetch` -- `REPLICATOR_CONSUMER_NAME` — this worker's identity within the group; defaults to `replicator@`. Two workers must never share one — Redis tracks pending entries per consumer name, and a shared name makes independent `claim_stale` recovery impossible -- `REPLICATOR_CONSUMER_START_ID` — group start position; default `"$"` (new messages only), `"0"` drains the backlog. Applies **only at group creation** — once `replicator.fetch` exists, changing this also needs a manual `XGROUP SETID` -- `REPLICATOR_READ_BLOCK_MS` — blocking-read window; default `5000`. Bounds worst-case shutdown latency, so the unit's `TimeoutStopSec` must exceed it plus the handler budget **plus an in-flight sweep** — `asyncio.to_thread` puts the tree walk beyond cancellation, so SIGTERM waits it out -- `REPLICATOR_MIN_HOST_INTERVAL_SECONDS` — minimum spacing for a host with **no explicit policy**; default `1.0`. Since #19 this is the *fallback*, not the rule: the per-host numbers arrive on `content.fetch-policy` and an unknown, revoked, or not-yet-replayed host resolves here — never to unlimited, because a boot replay cannot tell a consumer whether the set it received is whole. 1.0 is Watcher's own `DEFAULT_MIN_INTERVAL`, chosen because it invents nothing. A wait ≤ `REPLICATOR_READ_BLOCK_MS` is slept through in the handler; a longer one raises `TransientFetchError` and parks the command, so the effective floor on a *parked* wait is `REPLICATOR_CLAIM_MIN_IDLE_MS`. **`0` no longer disables pacing outright** (#19 narrowed it): it is the fallback for unpublished hosts only, and a host with a policy is still paced by it — letting an env var veto a published value would invert the ownership split the charter settles. Capped at `3600`: past an hour the command parks and re-parks without ever dead-lettering (transient failures are exempt from the delivery ceiling) while the issuer's reaper concludes loss, so a fat-fingered extra zero should fail at startup rather than read as healthy. **It cannot be validated against what a producer might publish** — a published interval has no upper bound — so the strictness contract is enforced the only place it is knowable: a WARNING per host at apply time when a real policy turns out stricter than this number -- `REPLICATOR_CLAIM_MIN_IDLE_MS` — idle time before a pending entry may be reclaimed; default `60000`. Doubles as the retry cadence -- `REPLICATOR_MAX_DELIVERY_ATTEMPTS` — delivery ceiling for *unclassified* failures before DLQ; default `5`. Counted from XPENDING's delivery counter, which only advances on a reclaim ⇒ a bound in time, not retries -- `REPLICATOR_ERROR_BACKOFF_BASE_SECONDS` / `REPLICATOR_ERROR_BACKOFF_MAX_SECONDS` — backoff for a poll *cycle* that raised (broker outage); defaults `1.0` / `30.0`, escalating `base * 2**(n-1)` -- `REPLICATOR_MAX_CONSECUTIVE_CYCLE_FAILURES` — consecutive failed cycles before the worker exits so the unit restarts; default `20` (~8 min at the default backoff). Paired with the unit's `StartLimitIntervalSec=3600` / `StartLimitBurst=3` -- `REPLICATOR_DEDUPE_TTL_SECONDS` — lifetime of the `replicator:cmd:` dedupe key; default `86400` -- `REPLICATOR_LOG_LEVEL` — default `INFO`. Governs the **root** logger only, which is the whole tree for the worker. Under the dev server's `--log-config`, uvicorn's own `uvicorn` / `uvicorn.access` / `uvicorn.error` loggers are pinned `INFO` by `src/core/log_config.json` and do not follow it (nor did they under uvicorn's built-in config), so setting `WARNING` will not silence access lines; root itself is `INFO` from boot until the lifespan's `configure_logging()` applies this value -- `BUILD_ID` — git SHA stamped by the systemd unit's `ExecStartPre`; defaults to `"dev"` outside systemd +Every variable the service reads, with the reasoning behind each default: +[docs/DEPLOYMENT.md](docs/DEPLOYMENT.md). ## Bus Conventions Replicator is a **consumer** first. Follow the conventions co-core and the archiver producer established: -- **The issuer contract is written down and lives here.** `docs/contracts/content-fetch-issuer-contract.md` is the normative statement of what a `content.fetch` producer must do — per-occasion `command_id`, `url` is not a correlation key, persist the `command_id → domain` map before publishing, correlate idempotently, and keep a reaper as a backstop for the outcomes no fact can carry. Issuer-side repos (Watcher, Phase 4) link to it rather than copying it. Anything asserted there is asserted about this repo's code; change one, change both (#8). -- **What Replicator is allowed to become is also written down.** `docs/contracts/replicator-boundaries.md` is the sibling charter: mechanism to Replicator, policy to the issuer, config over the bus, and an inbound admin HTTP API rejected by name. Run its three tests against any proposed capability, field, or setting before writing code — a database, domain vocabulary, or a write route is reached one defensible step at a time, not in one commit. `tests/test_boundaries.py` enforces eight invariants in CI, including the one that catches the regression review misses: an AST scan of `src/` for domain nouns in identifiers *and* string literals. Known violation recorded and pinned rather than omitted — `blob_uri` is host-local `file://` (#7). Change the charter and the tests together (#12). -- **`content.blobs` carries both outcomes.** `blob_available` on success, `fetch_failed` on a command closed without bytes (#9, co-core cannobserv#270 — v0.7.2). One stream so an issuer's single consumer group sees either. The reason is named at the *raise site* (`PermanentFetchError.reason`), never recovered from a message string, because three unrelated permanent conditions share one exception type. Three rows stay DLQ-only and permanently silent, and only one of them for want of an id: a frame that did not decode, a command whose `command_id` is blank (refused before the fetch — an empty id would otherwise take the dedupe key `replicator:cmd:` and make every later blank-id command a silent no-op, CR #6), and a frame that decoded to a **non-command payload** — the last is unreportable not because it lacks a `command_id` but because any it carries is *another command's* (`BlobAvailableEvent`'s names one that succeeded), so a terminal fact keyed on it would contradict a fact the issuer already applied (CR #1). `_close` refuses a correlator-less report at the one choke point rather than at each call site. Non-terminal facts are deferred (#9 §3): the stream is broadcast and nothing trims it, so a fact per reclaim during an origin outage is unbounded growth. `src/worker/reporter.py`. -- **`blob_available` carries the fetch, not just the bytes.** Six optional fields beyond the blob itself (#10, cannobserv#271 + #279 — v0.7.5): `final_url`, `status_code`, `fetched_at`, `content_type_raw`, `etag`, `last_modified`. Each is what Replicator holds at publish time and a broadcast consumer cannot recover now that fetching lives here rather than in Watcher. **`None` means nobody said it** — never a stand-in: `final_url` is never backfilled from `command.url` (an issuer could no longer tell "landed where I asked" from "nobody knows"), and `content_type_raw` is never backfilled with `DEFAULT_MEDIA_TYPE` (which is the value a consumer reads as "unknown, guess from the URL"). `media_type` keeps its normalized semantics beside the raw channel; the two are not interchangeable. `fetched_at` is stamped where the fetch *returns*, not at publish — `occurred_at` under a reclaim is minutes late. `status_code` is always 2xx on this fact, so it distinguishes 200 from 203/206 and is not a success branch. The passthroughs are **dropped over `MAX_HEADER_VALUE_LENGTH`, never truncated**: these are origin-controlled strings on a stream nothing trims, and a truncated ETag replayed in an `If-None-Match` is a validator that can never match. `src/worker/handler.py::_passthrough`. -- **The failure fact is a seam, not a call in the loop.** `FailureReporter` is injected exactly as `Handler` is, so `loop.py` stays ignorant of `content.blobs` and `blobs_topic` stays a defaulted argument a live-broker test can move. The loop owns the *decision* (`terminal` is "did this hit the delivery ceiling", which only the loop knows); the reporter owns the *publish*. `_close()` publishes then dead-letters, so **fact-before-ack** cannot be got wrong one call site at a time — `dead_letter` acks inside itself, and a fact published after it is lost outright on a crash. A failed fact-publish is **swallowed**, deliberately asymmetric with the byte path's `_publish`: there raising prevents an orphan blob, here the DLQ entry is already the durable record and raising would burn the delivery ceiling to reach the same DLQ minutes later. -- **A command shapes its own fetch, and everything unsendable is refused rather than fixed.** `headers` and `timeout_seconds` (#11, cannobserv#272 — v0.7.3) reach the driver as `FetchContent.headers` / `.timeout`; omitted means the pre-#11 wire byte-for-byte. Header names are **lower-cased before the merge** — `AsyncFetchDriver` merges `{"user-agent": DEFAULT, **effect.headers}` case-*sensitively*, and httpx does not resolve the collision: an unfolded `User-Agent` puts **two** field lines on the wire, default first, for the origin to disambiguate (measured, not inferred). That is exactly the fingerprint-continuity case Watcher needs at cutover. Refusals are **`PermanentFetchError(INVALID_REQUEST_OPTIONS)` raised before the fetch**: hop-by-hop and httpx-derived names (`host`, `content-length`, `proxy-*`, …), non-token names (padding included — OWS is a *value* rule, not a name one), values outside **printable US-ASCII**, case-collisions, over `MAX_REQUEST_HEADERS`/`MAX_REQUEST_HEADER_BYTES`, and a timeout that is non-finite, ≤ 0, or over `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS`. Refused, never stripped or clamped — same argument as dropping an over-long passthrough rather than truncating it: a change to the fetch the issuer cannot see is one it cannot account for in its own fingerprints. **The value charset is pinned to what httpx can actually send, not to RFC 9110** — see `_HEADER_VALUE`'s comment for why each edge sits where it does (CR #1). The rule the guard exists to enforce: a value httpx refuses does not fail as a classified fetch error, so an unguarded one closes the command under the wrong reason or retries forever. `tests/worker/test_handler_request_options.py` walks the whole single-byte range against `httpx.Request` so the two cannot drift again. Validation runs **ahead of the storage ceiling** so a permanently-bad command does not park in the PEL waiting for a sweep. Header **names only** are logged, never values. `src/worker/handler.py::_request_options`. -- **Politeness is enforced here and decided elsewhere, and since #19 the numbers actually arrive.** `src/worker/pacing.py` holds a host → last-request map in memory (derived, bounded, rebuildable — one of the three state shapes the boundaries charter permits) and reports a wait; `handler.py::_pace` spends it. **Two ways to spend it, split by duration, because neither works alone on a serial consume path**: a wait ≤ `REPLICATOR_READ_BLOCK_MS` is slept through, a longer one raises `TransientFetchError` and parks the command for `claim_stale`. Park-only was the obvious design and is wrong by 60× — a parked wait cannot be shorter than `REPLICATOR_CLAIM_MIN_IDLE_MS` (60 s) while the normal interval is 1 s, so every host would have been paced at 1/60th of today's rate, silently and in the safe direction. Sleep-only holds every *other* host's commands, and a SIGTERM, behind one origin's politeness. Transient in both directions so being polite can never burn the delivery ceiling. The pacer is built from settings when not injected — the seam fails **open**, and a byte path that quietly stopped pacing is indistinguishable from one that is working. Only a request that actually goes out calls `record()`: stamping a parked attempt would space the origin from requests it never received. **Keyed on the host asked for, not the host reached** — httpx follows redirects inside the driver, so URLs funnelling into one portal hit it at N× the intended rate; recorded as a known limitation in the charter rather than fixed here, because the fix breaks "one request, one record" — **#19 did not resolve it**, it only made the fix more defensible. Signal: `paced_seconds` on the byte path's success line (per-fetch, correlate with `duration_ms`), `tracked_hosts` on `_pace`'s own INFO line, which fires only when a wait was actually spent. The pacer resolves each host's interval through the `policy` seam — a bare callable, not the map, so `handler.py` stays ignorant of `content.fetch-policy` the way `loop.py` stays ignorant of `content.blobs` (#12, #19, watcher#245, cannobserv#285). - -- **`content.fetch-policy` is the third stream kind, and it is read groupless.** `src/worker/policy.py` replays it from `0-0` at boot and tails it thereafter; `FetchPolicyMap` is the state, `run_policy_reader` the poll loop, a peer of the consume loop and the retention sweep in `_run_until_first_exit`. Hyphen, not a third dot segment — `content.fetch.policy` collides with the `.dlq` derivation of the command stream, so use `streams.CONTENT_FETCH_POLICY`. **No consumer group**: every worker needs every message, and a group here grows a PEL nothing drains — hence no `ack` and no DLQ either, and a frame that will never decode is skipped by forcing the cursor (`seek`). Reads are **`count=1` throughout**, deliberately: `AsyncBusTailReader` advances its cursor only on a fully-decoded batch, so recovering from a poison frame at `count>1` means draining the well-formed prefix at `count=1` *before* seeking past it (`seek` only moves forward) — one message per host per republish makes the extra round trips free and deletes that ordering entirely. **`AsyncBusTailReader.replay()` is unusable and `PolicyReader` deliberately omits it** (#19 CR #1): it accumulates across many `read` calls and returns the list only on a clean finish, so *any* raise part-way through discards everything it read while the cursor has already moved — a poison frame at position *k* silently loses the *k−1* policies ahead of it, permanently. Drive `read` yourself and apply each batch as it arrives. It is the obvious API and its name says exactly what a caller wants, which is why this warning is here rather than only in the docstring. Recovery is bounded (`MAX_POISON_SKIPS`, shared with `loop.py`) because an anomaly clears the outage counter, so a run of them would otherwise spin at broker round-trip speed with backoff permanently disarmed; and the boot replay takes the **stop event** so a SIGTERM during an untrimmed stream's replay is not ignored until it finishes. **Replay runs synchronously before the consume loop starts**, or the worker's opening commands are paced against an empty map; a failed replay is absorbed rather than fatal, because the cursor advanced only over what decoded and the tail drains the rest. Four ways to apply a message wrongly and silently — branch on `revoked` before the interval (`min_interval_seconds` is `None` on a tombstone), treat `0.0` as a value rather than an absence, `isinstance`-check the payload (`from_wire`'s table is global, so a `blob_available` here decodes into the wrong model rather than raising), and guard on `occurred_at` per host so a stale full-set republish cannot revert a newer change. The reader **absorbs its own failures** like `run_sweeper` does: politeness is not load-bearing for correctness, and a broker that is genuinely gone surfaces through the consume loop, which has the delivery obligations. Its blocking read uses `REPLICATOR_READ_BLOCK_MS`, the same window the consume loop uses and concurrently with it, so `TimeoutStopSec` gains no term (#19, cannobserv#285 — v0.7.7). -- **The timeout ceiling and the unit's `TimeoutStopSec` are one decision.** A command's own timeout replaced the driver's fixed 30 s as the handler's worst-case budget, and SIGTERM waits out the message in flight, so `TimeoutStopSec` must exceed `REPLICATOR_READ_BLOCK_MS` + `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` + an in-flight sweep. `tests/test_deploy.py` enforces the first two terms; the third is the margin. -- **At-least-once ⇒ idempotent.** Two idempotency keys, two levels: the **command** dedupes on `command_id`, the **fact** on `content_fingerprint` — and `fetch_failed` on neither, keyed `command_id:occurred_at` so a consumer's dedup-on-key cannot collapse a multi-emission sequence and drop the terminal event. The two serve different purposes and are not interchangeable — the fingerprint is *storage* identity, `command_id` is *correlation* identity, and a consumer deduping its inbox on the fingerprint silently loses the second of two commands that fetched identical bytes. Content-addressed storage makes re-storing identical bytes a no-op regardless. -- **Consumers must be idempotent; producers own the outbox.** The cluster split (parent strategy, "Delivery + correctness") assigns the transactional outbox to producers with a DB system of record. Replicator has none — its durable record of intent is the consumer group's PEL, recovered via `claim_stale`. Do not add a Postgres outbox to the consume path. -- **Validation posture:** use the canonical `extra="ignore"` models; **branch on `schema_version` before destructuring**; tolerate additive producer fields. Never use the strict `*Emit` classes on the consume path. -- **Batch-poison caveat:** `AsyncBusConsumer.read(count>1)` raises `BusMessageAnomaly` on a malformed frame *before* returning the well-formed ones in the batch. Read `count=1`, or catch the anomaly and route via `dead_letter`. `from_wire` is deliberately fail-loud. -- **DLQ is a shipped seam, not a TODO:** `dead_letter(message_id, fields)` copies the frame to `.dlq` and acks the original. Deterministic failure ⇒ DLQ; transient failure ⇒ retry. -- **A frame that fails to decode has no fields.** `from_wire` raises from *inside* `read`/`claim_stale`, so the anomaly carries `topic` + `message_id` only — but `dead_letter` XADDs the fields it is given and `XADD` rejects an empty map. Re-read the raw frame by id (`XRANGE topic id id`) and fall back to a synthesized record when the entry has been trimmed. `src/worker/loop.py::dead_letter_anomaly`. -- **`from_wire`'s dispatch table is global.** A `blob_available` frame XADDed to `content.fetch` decodes cleanly into the wrong model rather than raising — `isinstance`-check the payload before destructuring. -- **`claim_stale` is the retry path, not just crash recovery.** A transiently-failed message is left unacked and comes back through the same reclaim, so retry cadence = `REPLICATOR_CLAIM_MIN_IDLE_MS`. Call it with `count=1`: XAUTOCLAIM transfers ownership and resets the idle clock on every entry it returns *before* co-core decodes them, and it restarts at `0-0` each call, so a poison entry jams recovery permanently unless it is DLQ'd first. -- **Retry accounting is XPENDING's `times_delivered`**, not a side counter. It only advances on a reclaim. -- **A consumer appears in `XINFO CONSUMERS` only after its first *delivered* message.** An empty poll registers nothing, so an absent consumer entry is not evidence a worker is down — a liveness check built on it reports every idle worker as dead. Recovery is unaffected: `claim_stale` reclaims by group and idle time, not by a pre-existing consumer entry. -- **A failing *message* and a failing *cycle* are different.** `process_message` decides a message's fate; a broker refusing reads/acks/DLQ writes is `run_loop`'s problem — it backs off (`REPLICATOR_ERROR_BACKOFF_BASE_SECONDS` → `_MAX_SECONDS`) and retries, then re-raises after `REPLICATOR_MAX_CONSECUTIVE_CYCLE_FAILURES` so a permanently wrong `REPLICATOR_REDIS_URL` surfaces as a restart instead of a worker that looks alive while doing nothing. The unit's `StartLimitIntervalSec` is sized against that ceiling — change one, revisit the other. -- **Bus clients are injection-only** — the co-core driver never opens or closes the `redis.asyncio.Redis` client. The worker owns one for its lifetime. -- **Store, then publish — never the reverse.** A crash between the two must not leave a `blob_available` pointing at bytes that are not there: a consumer would read the fact, fail to open the blob, and have no way to ask again. The opposite gap (stored bytes, no fact) repairs itself — the message stays unacked and the reclaim re-runs a handler that content-addressed storage makes a no-op. -- **A blob is `file://///.bin`.** Sharded two levels to bound directory fan-out; the extension is a constant `.bin`, **never** derived from `media_type` — identical octets can arrive under different Content-Types, and two paths for one fingerprint would defeat `exists()` as a short-circuit. Writes go through a temp file + `os.replace`: presence at a content-addressed path is what readers take as proof the bytes are complete. Design: `docs/plans/2026-07-31-replicator-mvp-open-questions-design.md`. -- **Blob modes are set on creation only.** Files land at `0644`, directories the worker creates at `0755` — both by explicit `chmod`, since `mkstemp` creates at `0600` and `mkdir`'s mode is masked by the umask. A directory that **already exists is never re-chmod'd**: it belongs to whoever provisioned it, and `chmod` on an unowned-but-writable mount raises `EPERM`. The cost is a silent trap — a `0700` level anywhere in the chain stores and publishes normally while no other service can open the `blob_uri` — so `warn_if_unreachable` walks `blob_dir` **and every parent** at startup and names each blocking level. Traversal needs `+x` all the way up, and the likeliest mistake is a restrictive parent over a fine leaf. `src/storage/local.py::ensure_directory`, `src/worker/main.py::warn_if_unreachable`. -- **Fetch outcomes carry the loop's vocabulary, not HTTP's.** 5xx / 408 / 429 ⇒ `TransientFetchError`; every other non-2xx (including a body-less 304, which passes `is_success`) ⇒ `PermanentFetchError`. httpx's exception hierarchy is **disjoint from the builtin `ConnectionError`/`TimeoutError`** the loop already treats as transient, so `src/worker/handler.py` maps it explicitly — leaving it unmapped would burn the delivery ceiling on an origin outage. -- **`occurred_at` is enforced tz-aware UTC on every payload** since co-core v0.7.2 (cannobserv#273). Naive is rejected fail-loud rather than assumed UTC; aware non-UTC is normalized. Load-bearing beyond tidiness — `isoformat()` is half `fetch_failed`'s envelope key, and a naive value would serialize without an offset. Issuer-visible: a naive `occurred_at` now fails `from_wire` and dead-letters as an anomaly. -- **`from_wire`'s topic and message_id are keyword-only** — `from_wire(fields, topic=..., message_id=...)`. The founding plan's API table showed them positionally. -- `sha256` lives at `co_core.pure.util.hashing`, not `co_core.pure.extract` (which carries `simhash`, `Chunk`, and the parsers). Import parsers from submodules — they are not re-exported from `__init__`. - -- **Nothing but the seed script writes to `content.fetch`.** `scripts/seed_fetch.py` requires `--redis-url` and `--topic` explicitly and additionally requires `--production` for the one combination the live worker consumes (db 0 **and** `content.fetch`) — a frame there is fetched for real. Its `--watch` reads the fact stream with a plain `XREAD` and never joins a group: a group left by an operator tool accumulates a PEL nothing drains. The stream watched follows `--topic` (`content.blobs` for `content.fetch`, `.blobs` otherwise), so a scratch seed does not sit watching production's facts. -- **The command and fact topics are defaulted arguments, not settings.** `build_consumer(..., topic=)` and `build_handler(..., blobs_topic=)` exist so a live-broker test can work on `replicator.itest.*` streams. No deployment wants a different stream, and configuring it would put the production one an operator's typo away. - -### Retention - -`docs/plans/2026-07-31-replicator-mvp-open-questions-design.md` §4 scope-cut retention; #5 settles it. Replicator is the producer in archiver's temp-cache protocol, where **the producer cleans up**. - -- **The TTL runs from last reference, not first store.** `store` short-circuits on an existing content-addressed path but its caller publishes a fresh `blob_available` either way, so a re-fetch of unchanged bytes would otherwise announce a blob already partway through its TTL. `LocalBlobStore` therefore `os.utime`s on the short-circuit branch, swallowing `ENOENT` — the sweep can unlink between the existence check and the touch, and the fallout of that race is a `blob_uri` that fails to open, not a dead-lettered command. -- **The blob tree holds three populations, and they are not interchangeable.** Finished blobs (`//.bin`) reap on the TTL; in-flight temporaries (`...tmp`) are **not garbage** — reaping one makes the writer's `os.replace` fail with `ENOENT` and dead-letters a good command — so the sweep matches `*.bin`, never `iterdir()`, and ages temps out on their own much longer grace. Empty shard directories go **last**, by `rmdir` only, whose refusal to touch a non-empty directory is the safety property. -- **The ceiling is backpressure, not a faster clock.** Over `REPLICATOR_BLOB_MAX_TOTAL_BYTES` the byte path raises `TransientFetchError` *before* fetching, so the command stays in the PEL and returns via `claim_stale` once a sweep frees space. What it measures is everything the tree holds — surviving blobs **and** the temporaries the sweep is waiting out, since a crash loop fills the disk with debris the ceiling would otherwise not see. The per-population counts stay split in the sweep log so a rising temp count cannot hide inside a healthy blob one. Reaping a blob still inside its TTL to make room would convert a local disk problem into a `blob_uri` another repo cannot open — the one failure mode with no local symptom. -- **One `BlobUsage`, two writers.** The sweep's `observe` is the measured total; the byte path's `add` is the estimate between sweeps, because a burst can cross the ceiling long before the tree is walked again. Wiring the two halves to separate instances leaves both individually correct and the guard permanently unreachable — `tests/worker/test_main.py` pins the identity. -- **Orphans are recorded where they are exact.** A publish that fails after the store leaves bytes with no fact and no `command_id`, invisible to any query starting from `content.blobs`. `src/worker/handler.py::_publish` logs the fingerprint at that moment and re-raises untouched; the sweep then treats orphans as ordinary aged blobs. Reconciling the tree against `content.blobs` instead would make a *delete* decision depend on another service's stream-trimming policy. -- **The sweep runs in the worker, not a systemd timer.** A timer survives a crashed worker, but a worker that is not running is not writing blobs either. It rides the same stop event as the consume loop (`src/worker/loop.py::park`) and walks the tree via `asyncio.to_thread`, so retention never becomes a source of consume-path latency. A failed sweep is absorbed and retried next cycle — retention is not load-bearing for correctness, and the ceiling is the guard for a tree that cannot be reaped. - -**Testing the bus.** `tests/conftest.py` ships a `fake_redis` fixture (fakeredis, Streams-capable) — consumer-group behaviour is testable without a broker, and assertions should read the broker's own view (`xinfo_groups` / `xinfo_consumers`) rather than co-core's private attributes, which are not a stable contract. Anything that genuinely needs the live Archiver-operated Redis goes behind `@pytest.mark.integration` and is excluded by default. - -**Where fakeredis diverges.** It is sound for consumer-group *mechanics* — what state a command leaves behind — but diverges on *lifecycle* and *blocking* semantics: it registers a consumer on an empty `XREADGROUP` (real Redis waits for a delivery, GH #3) and it ignores `block` (worked around by `IDLE_SLEEP_SECONDS` in `src/worker/loop.py`). Rule of thumb: an assertion about **what state results** is safe against the fake; an assertion about **when Redis does something** needs a live broker. Both divergences were found by running against the real server, not by the suite. - -Live-broker tests use the `real_redis` fixture (`tests/conftest.py`), which connects to `REPLICATOR_TEST_REDIS_URL` (default `redis://localhost:6379/15`), skips when nothing answers (an *auth* failure re-raises — a misconfigured broker must not pass as a skip), expires stray `replicator.itest.*` keys from crashed runs once per session, and refuses db 0 outright — db 0 carries the live `content.fetch` stream that the running `replicator.service` consumes, so a test frame written there would be fetched for real. Confine such tests to scratch stream keys via the `scratch_topic` fixture (`tests/worker/conftest.py`), whose teardown also removes `.dlq`; the database guard is the backstop, not the plan. - -**One namespace the sweeper cannot reach.** `process_message` writes `replicator:cmd:`, a constant prefix outside `replicator.itest.*`, so an end-to-end test deletes its own keys via the `dedupe_keys` fixture and shortens their TTL. `test_an_end_to_end_run_only_creates_predictable_keys` asserts the whole promise: every key a run creates is either an itest stream or a dedupe key. +- **At-least-once ⇒ idempotent.** The command dedupes on `command_id`, the fact on + `content_fingerprint`, and `fetch_failed` on neither — storage identity and + correlation identity are not interchangeable. +- **Store, then publish — never the reverse.** A fact pointing at bytes that are + not there is unrepairable by the consumer; stored bytes with no fact repair + themselves on the reclaim. +- **Read `count=1`.** `AsyncBusConsumer.read(count>1)` raises on a malformed frame + *before* returning the well-formed ones, and `claim_stale` at `count>1` lets a + poison entry jam recovery permanently. +- **`from_wire` is fail-loud and its dispatch table is global** — `isinstance`-check + every decoded payload before destructuring. Use the canonical `extra="ignore"` + models on the consume path, never the strict `*Emit` classes, and branch on + `schema_version` first. +- **Deterministic failure ⇒ DLQ; transient failure ⇒ retry.** `dead_letter` acks + inside itself, so a fact is published *before* it. Retry cadence is + `REPLICATOR_CLAIM_MIN_IDLE_MS`; a failing *cycle* is `run_loop`'s problem, not + the message's. +- **Consumers must be idempotent; producers own the outbox.** Replicator has no DB + — its durable record of intent is the consumer group's PEL. Do not add a + Postgres outbox to the consume path. +- **Three stream kinds, three sets of rules.** `content.fetch` is the command + stream, `content.blobs` carries both outcomes (`blob_available` and + `fetch_failed`), and `content.fetch-policy` is read **groupless** — no group, no + ack, no DLQ. Read [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) before changing + what any of them carries. +- **Nothing but the seed script writes to `content.fetch`.** `scripts/seed_fetch.py` + requires `--production` for the one combination the live worker consumes — a + frame there is fetched for real. +- **Two normative contracts bound the wire and the roadmap**, both under + `docs/contracts/` and linked from sibling repos. `tests/test_boundaries.py` + enforces eight charter invariants in CI; change a charter and its tests + together. + +Blob paths, modes, and the retention sweep: [docs/STORAGE.md](docs/STORAGE.md). +Fakeredis's divergences from the live broker, and the keys an integration run may +touch: [docs/TESTING.md](docs/TESTING.md). ## Common Commands @@ -291,11 +205,8 @@ logger = get_logger(__name__) ``` Entry points only: `configure_logging()` is called once inside the FastAPI `lifespan` or the worker's `run()`. Never in library modules. -**One formatter, two installers.** `build_json_formatter()` is the single definition of the JSON schema (`timestamp`, `level`, `logger`, `message`). `configure_logging()` installs it on the root logger; `src/core/log_config.json` names the *same factory* through dictConfig's `"()"` key, so there is no second fmt string to drift. The dev server must be launched with `--log-config src/core/log_config.json` — uvicorn's `uvicorn` / `uvicorn.access` / `uvicorn.error` loggers ship with `propagate=False` and their own plain-text handlers, so a root-only config never reaches them and the output is half JSON, half plain text. **The worker runs no uvicorn**, so `replicator.service` needs no `--log-config`; its `ExecStart` is `python -m src.worker.main` and `configure_logging()` is the whole story there (#14). - -**Not everything in the journal is JSON.** The claim above is about the *app's own records*. `replicator.service`'s wheelhouse-sync `ExecStartPre` writes a **plain-text** line to journald on every start — `wheelhouse in sync: N downloaded, M already present -> …` on the happy path, `error: could not sync gs://…` on the non-fatal failure path (stderr, which journald captures the same way) — and that is by design, not a gap: it runs `uv run --no-project` — before the deploy's `uv sync`, in an environment holding `google-cloud-storage` and nothing else — so it cannot import `build_json_formatter()` and making it emit JSON would mean a hand-maintained second copy of the schema in the one file that structurally cannot single-source it. A shipper reading journald natively is unaffected (the message is a field alongside `_SYSTEMD_UNIT` / `_PID`); a pipeline that `json.loads` every `MESSAGE` must tolerate these two lines — the failure one especially, since it appears exactly when something is already wrong (#15, skills#83). - -**The colour strip is a filter on the loggers, deliberately.** uvicorn attaches an ANSI-coloured duplicate of each lifecycle message as `extra={"color_message": ...}`, and every extra reaches the JSON payload. `ColorMessageFilter` deletes it from the record at its source — not via the formatter's `reserved_attrs`, and not on the stdout handler. Both alternatives scope the fix to *this* sink: a handler that builds its payload from the record's `__dict__` rather than a `logging.Formatter` resurrects the field, and OpenTelemetry's `LoggingHandler` is exactly that (its own reserved list does not cover `color_message`). Mutating the record once means the strip survives a sink swap with no failing test to warn you it had stopped working. `tests/core/test_logging.py` pins the filter's placement, not just its effect. +The logging stack — one formatter, two installers, and the two journald lines that +are deliberately not JSON: [docs/STYLE.md](docs/STYLE.md). **Date & Time:** - All UTC @@ -307,3 +218,15 @@ Entry points only: `configure_logging()` is called once inside the FastAPI `life - Test structure mirrors source (`src/foo.py` → `tests/test_foo.py`). A module whose tests outgrow one file splits by concern, not by helper: `tests/worker/test_loop_dlq.py`, `test_loop_recovery.py`, … with the shared wiring in that package's `conftest.py`. Concern is the default axis; **environment** is the one exception — tests needing a live broker split off with an `_integration` suffix (`tests/worker/test_main_integration.py`), so the filename says what the marker enforces - Explicit imports only - Small, focused functions + +## Detail Docs + +- [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) — module-by-module layout and every bus contract's reasoning; read before changing what a stream carries +- [docs/STORAGE.md](docs/STORAGE.md) — blob paths and modes, the three populations under `REPLICATOR_BLOB_DIR`, TTL and ceiling semantics +- [docs/DEPLOYMENT.md](docs/DEPLOYMENT.md) — VM topology, ports, the systemd unit's lifecycle, and every environment variable the service reads +- [docs/TESTING.md](docs/TESTING.md) — where fakeredis diverges from the live broker, and which keys an integration run may create +- [docs/STYLE.md](docs/STYLE.md) — the logging stack: formatter, installers, and the non-JSON journald lines +- [docs/COMMANDS.md](docs/COMMANDS.md) — every runnable command, with flags +- [docs/SKILLS.md](docs/SKILLS.md) — vendored skill inventory and refresh procedure +- [docs/contracts/content-fetch-issuer-contract.md](docs/contracts/content-fetch-issuer-contract.md) — what a `content.fetch` producer must do; normative, linked from issuer repos +- [docs/contracts/replicator-boundaries.md](docs/contracts/replicator-boundaries.md) — what Replicator may become; run its three tests against any proposed capability diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md new file mode 100644 index 0000000..c49100f --- /dev/null +++ b/docs/ARCHITECTURE.md @@ -0,0 +1,74 @@ +# Replicator Architecture + +Module layout and the bus contracts the worker consumes and publishes. +`AGENTS.md` keeps the rules needed on nearly every task; the reasoning that makes +each one non-negotiable lives here. Blob-tree and retention rules are in +[STORAGE.md](STORAGE.md). + +## Project Layout + +``` +src/worker/ — Bus consumer; the primary process +src/worker/main.py — Entry point: client lifetime, consumer group, signals +src/worker/loop.py — The consume path: poll → dispatch → ack, DLQ, dedupe, recovery +src/worker/handler.py — The byte path behind the Handler seam: fetch → fingerprint → store → publish +src/worker/reporter.py — The failure fact behind the FailureReporter seam: fetch_failed on content.blobs +src/worker/retention.py — The sweep task: cadence, usage accounting, ceiling reporting +src/worker/pacing.py — Per-host request spacing; the mechanism half of politeness (#12) +src/worker/policy.py — The content.fetch-policy consumer: the map, and the groupless tail (#19) +src/storage/ — Temp storage; BlobStore protocol + the local-FS backend +src/storage/base.py — BlobStore protocol (store / exists / open) +src/storage/local.py — Content-addressed local backend; file:// URIs, sharded paths +src/storage/sweeper.py — Retention: TTL reap, stale temps, empty shards; the measured size +src/api/ — FastAPI app (/health only; not part of the MVP loop) +src/api/main.py — App factory, lifespan, router registration +src/core/ — Shared domain logic, logging, config +src/core/errors.py — TransientFetchError / PermanentFetchError + FailureReason (handler failure vocabulary) +src/core/logging.py — build_json_formatter() + ColorMessageFilter + configure_logging() + get_logger() +src/core/log_config.json — uvicorn --log-config; routes uvicorn's own loggers through that formatter +src/core/config.py — Settings / env access (see Environment Variables) +scripts/ — sync_wheelhouse.py, check_redis_floor.sh, seed_fetch.py +scripts/seed_fetch.py — the MVP command issuer; publishes content.fetch, --watch tails the facts +tests/ — Mirrors src/ structure; integration tests in `@pytest.mark.integration` +docs/ — Reference docs (COMMANDS, SKILLS) +docs/contracts/ — Normative contracts, linked to from sibling repos: the issuer-facing half of the wire, and the boundaries charter (what Replicator may become) +docs/plans/ — Implementation plans +deploy/ — Systemd unit + deployment config +.wheelhouse/ — Local mirror of the private cannobserv index (git-ignored except .gitkeep) +``` + +## Bus Conventions + +Replicator is a **consumer** first. Follow the conventions co-core and the archiver producer established: + +- **The issuer contract is written down and lives here.** `docs/contracts/content-fetch-issuer-contract.md` is the normative statement of what a `content.fetch` producer must do — per-occasion `command_id`, `url` is not a correlation key, persist the `command_id → domain` map before publishing, correlate idempotently, and keep a reaper as a backstop for the outcomes no fact can carry. Issuer-side repos (Watcher, Phase 4) link to it rather than copying it. Anything asserted there is asserted about this repo's code; change one, change both (#8). +- **What Replicator is allowed to become is also written down.** `docs/contracts/replicator-boundaries.md` is the sibling charter: mechanism to Replicator, policy to the issuer, config over the bus, and an inbound admin HTTP API rejected by name. Run its three tests against any proposed capability, field, or setting before writing code — a database, domain vocabulary, or a write route is reached one defensible step at a time, not in one commit. `tests/test_boundaries.py` enforces eight invariants in CI, including the one that catches the regression review misses: an AST scan of `src/` for domain nouns in identifiers *and* string literals. Known violation recorded and pinned rather than omitted — `blob_uri` is host-local `file://` (#7). Change the charter and the tests together (#12). +- **`content.blobs` carries both outcomes.** `blob_available` on success, `fetch_failed` on a command closed without bytes (#9, co-core cannobserv#270 — v0.7.2). One stream so an issuer's single consumer group sees either. The reason is named at the *raise site* (`PermanentFetchError.reason`), never recovered from a message string, because three unrelated permanent conditions share one exception type. Three rows stay DLQ-only and permanently silent, and only one of them for want of an id: a frame that did not decode, a command whose `command_id` is blank (refused before the fetch — an empty id would otherwise take the dedupe key `replicator:cmd:` and make every later blank-id command a silent no-op, CR #6), and a frame that decoded to a **non-command payload** — the last is unreportable not because it lacks a `command_id` but because any it carries is *another command's* (`BlobAvailableEvent`'s names one that succeeded), so a terminal fact keyed on it would contradict a fact the issuer already applied (CR #1). `_close` refuses a correlator-less report at the one choke point rather than at each call site. Non-terminal facts are deferred (#9 §3): the stream is broadcast and nothing trims it, so a fact per reclaim during an origin outage is unbounded growth. `src/worker/reporter.py`. +- **`blob_available` carries the fetch, not just the bytes.** Six optional fields beyond the blob itself (#10, cannobserv#271 + #279 — v0.7.5): `final_url`, `status_code`, `fetched_at`, `content_type_raw`, `etag`, `last_modified`. Each is what Replicator holds at publish time and a broadcast consumer cannot recover now that fetching lives here rather than in Watcher. **`None` means nobody said it** — never a stand-in: `final_url` is never backfilled from `command.url` (an issuer could no longer tell "landed where I asked" from "nobody knows"), and `content_type_raw` is never backfilled with `DEFAULT_MEDIA_TYPE` (which is the value a consumer reads as "unknown, guess from the URL"). `media_type` keeps its normalized semantics beside the raw channel; the two are not interchangeable. `fetched_at` is stamped where the fetch *returns*, not at publish — `occurred_at` under a reclaim is minutes late. `status_code` is always 2xx on this fact, so it distinguishes 200 from 203/206 and is not a success branch. The passthroughs are **dropped over `MAX_HEADER_VALUE_LENGTH`, never truncated**: these are origin-controlled strings on a stream nothing trims, and a truncated ETag replayed in an `If-None-Match` is a validator that can never match. `src/worker/handler.py::_passthrough`. +- **The failure fact is a seam, not a call in the loop.** `FailureReporter` is injected exactly as `Handler` is, so `loop.py` stays ignorant of `content.blobs` and `blobs_topic` stays a defaulted argument a live-broker test can move. The loop owns the *decision* (`terminal` is "did this hit the delivery ceiling", which only the loop knows); the reporter owns the *publish*. `_close()` publishes then dead-letters, so **fact-before-ack** cannot be got wrong one call site at a time — `dead_letter` acks inside itself, and a fact published after it is lost outright on a crash. A failed fact-publish is **swallowed**, deliberately asymmetric with the byte path's `_publish`: there raising prevents an orphan blob, here the DLQ entry is already the durable record and raising would burn the delivery ceiling to reach the same DLQ minutes later. +- **A command shapes its own fetch, and everything unsendable is refused rather than fixed.** `headers` and `timeout_seconds` (#11, cannobserv#272 — v0.7.3) reach the driver as `FetchContent.headers` / `.timeout`; omitted means the pre-#11 wire byte-for-byte. Header names are **lower-cased before the merge** — `AsyncFetchDriver` merges `{"user-agent": DEFAULT, **effect.headers}` case-*sensitively*, and httpx does not resolve the collision: an unfolded `User-Agent` puts **two** field lines on the wire, default first, for the origin to disambiguate (measured, not inferred). That is exactly the fingerprint-continuity case Watcher needs at cutover. Refusals are **`PermanentFetchError(INVALID_REQUEST_OPTIONS)` raised before the fetch**: hop-by-hop and httpx-derived names (`host`, `content-length`, `proxy-*`, …), non-token names (padding included — OWS is a *value* rule, not a name one), values outside **printable US-ASCII**, case-collisions, over `MAX_REQUEST_HEADERS`/`MAX_REQUEST_HEADER_BYTES`, and a timeout that is non-finite, ≤ 0, or over `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS`. Refused, never stripped or clamped — same argument as dropping an over-long passthrough rather than truncating it: a change to the fetch the issuer cannot see is one it cannot account for in its own fingerprints. **The value charset is pinned to what httpx can actually send, not to RFC 9110** — see `_HEADER_VALUE`'s comment for why each edge sits where it does (CR #1). The rule the guard exists to enforce: a value httpx refuses does not fail as a classified fetch error, so an unguarded one closes the command under the wrong reason or retries forever. `tests/worker/test_handler_request_options.py` walks the whole single-byte range against `httpx.Request` so the two cannot drift again. Validation runs **ahead of the storage ceiling** so a permanently-bad command does not park in the PEL waiting for a sweep. Header **names only** are logged, never values. `src/worker/handler.py::_request_options`. +- **Politeness is enforced here and decided elsewhere, and since #19 the numbers actually arrive.** `src/worker/pacing.py` holds a host → last-request map in memory (derived, bounded, rebuildable — one of the three state shapes the boundaries charter permits) and reports a wait; `handler.py::_pace` spends it. **Two ways to spend it, split by duration, because neither works alone on a serial consume path**: a wait ≤ `REPLICATOR_READ_BLOCK_MS` is slept through, a longer one raises `TransientFetchError` and parks the command for `claim_stale`. Park-only was the obvious design and is wrong by 60× — a parked wait cannot be shorter than `REPLICATOR_CLAIM_MIN_IDLE_MS` (60 s) while the normal interval is 1 s, so every host would have been paced at 1/60th of today's rate, silently and in the safe direction. Sleep-only holds every *other* host's commands, and a SIGTERM, behind one origin's politeness. Transient in both directions so being polite can never burn the delivery ceiling. The pacer is built from settings when not injected — the seam fails **open**, and a byte path that quietly stopped pacing is indistinguishable from one that is working. Only a request that actually goes out calls `record()`: stamping a parked attempt would space the origin from requests it never received. **Keyed on the host asked for, not the host reached** — httpx follows redirects inside the driver, so URLs funnelling into one portal hit it at N× the intended rate; recorded as a known limitation in the charter rather than fixed here, because the fix breaks "one request, one record" — **#19 did not resolve it**, it only made the fix more defensible. Signal: `paced_seconds` on the byte path's success line (per-fetch, correlate with `duration_ms`), `tracked_hosts` on `_pace`'s own INFO line, which fires only when a wait was actually spent. The pacer resolves each host's interval through the `policy` seam — a bare callable, not the map, so `handler.py` stays ignorant of `content.fetch-policy` the way `loop.py` stays ignorant of `content.blobs` (#12, #19, watcher#245, cannobserv#285). + +- **`content.fetch-policy` is the third stream kind, and it is read groupless.** `src/worker/policy.py` replays it from `0-0` at boot and tails it thereafter; `FetchPolicyMap` is the state, `run_policy_reader` the poll loop, a peer of the consume loop and the retention sweep in `_run_until_first_exit`. Hyphen, not a third dot segment — `content.fetch.policy` collides with the `.dlq` derivation of the command stream, so use `streams.CONTENT_FETCH_POLICY`. **No consumer group**: every worker needs every message, and a group here grows a PEL nothing drains — hence no `ack` and no DLQ either, and a frame that will never decode is skipped by forcing the cursor (`seek`). Reads are **`count=1` throughout**, deliberately: `AsyncBusTailReader` advances its cursor only on a fully-decoded batch, so recovering from a poison frame at `count>1` means draining the well-formed prefix at `count=1` *before* seeking past it (`seek` only moves forward) — one message per host per republish makes the extra round trips free and deletes that ordering entirely. **`AsyncBusTailReader.replay()` is unusable and `PolicyReader` deliberately omits it** (#19 CR #1): it accumulates across many `read` calls and returns the list only on a clean finish, so *any* raise part-way through discards everything it read while the cursor has already moved — a poison frame at position *k* silently loses the *k−1* policies ahead of it, permanently. Drive `read` yourself and apply each batch as it arrives. It is the obvious API and its name says exactly what a caller wants, which is why this warning is here rather than only in the docstring. Recovery is bounded (`MAX_POISON_SKIPS`, shared with `loop.py`) because an anomaly clears the outage counter, so a run of them would otherwise spin at broker round-trip speed with backoff permanently disarmed; and the boot replay takes the **stop event** so a SIGTERM during an untrimmed stream's replay is not ignored until it finishes. **Replay runs synchronously before the consume loop starts**, or the worker's opening commands are paced against an empty map; a failed replay is absorbed rather than fatal, because the cursor advanced only over what decoded and the tail drains the rest. Four ways to apply a message wrongly and silently — branch on `revoked` before the interval (`min_interval_seconds` is `None` on a tombstone), treat `0.0` as a value rather than an absence, `isinstance`-check the payload (`from_wire`'s table is global, so a `blob_available` here decodes into the wrong model rather than raising), and guard on `occurred_at` per host so a stale full-set republish cannot revert a newer change. The reader **absorbs its own failures** like `run_sweeper` does: politeness is not load-bearing for correctness, and a broker that is genuinely gone surfaces through the consume loop, which has the delivery obligations. Its blocking read uses `REPLICATOR_READ_BLOCK_MS`, the same window the consume loop uses and concurrently with it, so `TimeoutStopSec` gains no term (#19, cannobserv#285 — v0.7.7). +- **The timeout ceiling and the unit's `TimeoutStopSec` are one decision.** A command's own timeout replaced the driver's fixed 30 s as the handler's worst-case budget, and SIGTERM waits out the message in flight, so `TimeoutStopSec` must exceed `REPLICATOR_READ_BLOCK_MS` + `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` + an in-flight sweep. `tests/test_deploy.py` enforces the first two terms; the third is the margin. +- **At-least-once ⇒ idempotent.** Two idempotency keys, two levels: the **command** dedupes on `command_id`, the **fact** on `content_fingerprint` — and `fetch_failed` on neither, keyed `command_id:occurred_at` so a consumer's dedup-on-key cannot collapse a multi-emission sequence and drop the terminal event. The two serve different purposes and are not interchangeable — the fingerprint is *storage* identity, `command_id` is *correlation* identity, and a consumer deduping its inbox on the fingerprint silently loses the second of two commands that fetched identical bytes. Content-addressed storage makes re-storing identical bytes a no-op regardless. +- **Consumers must be idempotent; producers own the outbox.** The cluster split (parent strategy, "Delivery + correctness") assigns the transactional outbox to producers with a DB system of record. Replicator has none — its durable record of intent is the consumer group's PEL, recovered via `claim_stale`. Do not add a Postgres outbox to the consume path. +- **Validation posture:** use the canonical `extra="ignore"` models; **branch on `schema_version` before destructuring**; tolerate additive producer fields. Never use the strict `*Emit` classes on the consume path. +- **Batch-poison caveat:** `AsyncBusConsumer.read(count>1)` raises `BusMessageAnomaly` on a malformed frame *before* returning the well-formed ones in the batch. Read `count=1`, or catch the anomaly and route via `dead_letter`. `from_wire` is deliberately fail-loud. +- **DLQ is a shipped seam, not a TODO:** `dead_letter(message_id, fields)` copies the frame to `.dlq` and acks the original. Deterministic failure ⇒ DLQ; transient failure ⇒ retry. +- **A frame that fails to decode has no fields.** `from_wire` raises from *inside* `read`/`claim_stale`, so the anomaly carries `topic` + `message_id` only — but `dead_letter` XADDs the fields it is given and `XADD` rejects an empty map. Re-read the raw frame by id (`XRANGE topic id id`) and fall back to a synthesized record when the entry has been trimmed. `src/worker/loop.py::dead_letter_anomaly`. +- **`from_wire`'s dispatch table is global.** A `blob_available` frame XADDed to `content.fetch` decodes cleanly into the wrong model rather than raising — `isinstance`-check the payload before destructuring. +- **`claim_stale` is the retry path, not just crash recovery.** A transiently-failed message is left unacked and comes back through the same reclaim, so retry cadence = `REPLICATOR_CLAIM_MIN_IDLE_MS`. Call it with `count=1`: XAUTOCLAIM transfers ownership and resets the idle clock on every entry it returns *before* co-core decodes them, and it restarts at `0-0` each call, so a poison entry jams recovery permanently unless it is DLQ'd first. +- **Retry accounting is XPENDING's `times_delivered`**, not a side counter. It only advances on a reclaim. +- **A consumer appears in `XINFO CONSUMERS` only after its first *delivered* message.** An empty poll registers nothing, so an absent consumer entry is not evidence a worker is down — a liveness check built on it reports every idle worker as dead. Recovery is unaffected: `claim_stale` reclaims by group and idle time, not by a pre-existing consumer entry. +- **A failing *message* and a failing *cycle* are different.** `process_message` decides a message's fate; a broker refusing reads/acks/DLQ writes is `run_loop`'s problem — it backs off (`REPLICATOR_ERROR_BACKOFF_BASE_SECONDS` → `_MAX_SECONDS`) and retries, then re-raises after `REPLICATOR_MAX_CONSECUTIVE_CYCLE_FAILURES` so a permanently wrong `REPLICATOR_REDIS_URL` surfaces as a restart instead of a worker that looks alive while doing nothing. The unit's `StartLimitIntervalSec` is sized against that ceiling — change one, revisit the other. +- **Bus clients are injection-only** — the co-core driver never opens or closes the `redis.asyncio.Redis` client. The worker owns one for its lifetime. +- **Store, then publish — never the reverse.** A crash between the two must not leave a `blob_available` pointing at bytes that are not there: a consumer would read the fact, fail to open the blob, and have no way to ask again. The opposite gap (stored bytes, no fact) repairs itself — the message stays unacked and the reclaim re-runs a handler that content-addressed storage makes a no-op. + +- **Fetch outcomes carry the loop's vocabulary, not HTTP's.** 5xx / 408 / 429 ⇒ `TransientFetchError`; every other non-2xx (including a body-less 304, which passes `is_success`) ⇒ `PermanentFetchError`. httpx's exception hierarchy is **disjoint from the builtin `ConnectionError`/`TimeoutError`** the loop already treats as transient, so `src/worker/handler.py` maps it explicitly — leaving it unmapped would burn the delivery ceiling on an origin outage. +- **`occurred_at` is enforced tz-aware UTC on every payload** since co-core v0.7.2 (cannobserv#273). Naive is rejected fail-loud rather than assumed UTC; aware non-UTC is normalized. Load-bearing beyond tidiness — `isoformat()` is half `fetch_failed`'s envelope key, and a naive value would serialize without an offset. Issuer-visible: a naive `occurred_at` now fails `from_wire` and dead-letters as an anomaly. +- **`from_wire`'s topic and message_id are keyword-only** — `from_wire(fields, topic=..., message_id=...)`. The founding plan's API table showed them positionally. +- `sha256` lives at `co_core.pure.util.hashing`, not `co_core.pure.extract` (which carries `simhash`, `Chunk`, and the parsers). Import parsers from submodules — they are not re-exported from `__init__`. + +- **Nothing but the seed script writes to `content.fetch`.** `scripts/seed_fetch.py` requires `--redis-url` and `--topic` explicitly and additionally requires `--production` for the one combination the live worker consumes (db 0 **and** `content.fetch`) — a frame there is fetched for real. Its `--watch` reads the fact stream with a plain `XREAD` and never joins a group: a group left by an operator tool accumulates a PEL nothing drains. The stream watched follows `--topic` (`content.blobs` for `content.fetch`, `.blobs` otherwise), so a scratch seed does not sit watching production's facts. +- **The command and fact topics are defaulted arguments, not settings.** `build_consumer(..., topic=)` and `build_handler(..., blobs_topic=)` exist so a live-broker test can work on `replicator.itest.*` streams. No deployment wants a different stream, and configuring it would put the production one an operator's typo away. diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md new file mode 100644 index 0000000..15bbe80 --- /dev/null +++ b/docs/DEPLOYMENT.md @@ -0,0 +1,81 @@ +# Replicator Deployment + +Single-VM topology, the systemd unit's lifecycle, and every environment variable +the service reads. `AGENTS.md` keeps the two-env-file boundary and the restart +command; the per-variable reasoning is here. + +## Infrastructure + +Replicator shares the VM with archiver, watcher, and notifier: + +| Service | Framework | Port | Managed by | +|---|---|---|---| +| Worker (live) | asyncio bus consumer | — | `systemctl` (`replicator.service`) | +| API (dev) | FastAPI | 8041 | manual uvicorn | + +The worker binds no port. Port 8040 is reserved for Replicator's API should it ever be deployed; 8041 is the dev port. Neighbours: watcher 8000/8001, archiver 8020/8021, notifier 9000/9001. The exe.dev proxy transparently forwards ports 3000–9999; the dev server is reachable at `https://replicator.exe.xyz:8041/`. + +### Redis is Archiver-operated — Replicator connects, it does not run its own + +The Redis change bus is Archiver-operated cluster infrastructure (the shared VM's `redis-server.service`). Replicator is a **client**: never ship a broker, never claim ownership. + +**Redis ≥ 7.0 is Replicator-critical.** Replicator is the cluster's first user of `AsyncBusConsumer.claim_stale`, which reads `XAUTOCLAIM`'s three-element reply — the deleted-ids element added in Redis **server** 7.0. Below that, the crash-recovery path raises. `scripts/check_redis_floor.sh` guards this as an `ExecStartPre`. (The VM runs 7.0.15.) + +The **redis-py client** resolves `>=5,<8` transitively via `co-core-aio[bus]`. Don't re-pin it narrower. + +## Server Lifecycle + +**`replicator.service` runs the worker.** It binds no port, so there is no port to conflict over — but only one process should hold a given consumer name at a time. + +| Situation | Action | +|---|---| +| Code committed to main | `uv sync --frozen && sudo systemctl restart replicator` | +| Testing a worktree/branch | `uv run python -m src.worker.main` (set a distinct `REPLICATOR_CONSUMER_NAME`) | +| Debugging the live service | `sudo journalctl -u replicator -f` | +| After editing `deploy/replicator.service` | `sudo cp deploy/replicator.service /etc/systemd/system/ && sudo systemctl daemon-reload && sudo systemctl restart replicator` | +| After a co-core version bump | re-run `sync_wheelhouse.py`, then `uv sync` | + +`ExecStart` uses `--frozen --no-sync`, so dependency sync is a deploy step, not a service-start side effect. + +**`/etc/systemd/system/replicator.service` is a *copy*, not a symlink to `deploy/`.** So the `cp` above is load-bearing and `daemon-reload` alone silently does nothing — systemd re-reads the installed file, which is still the old one. The failure has no symptom at restart: the worker comes up on the new code under the *old* unit, and the mismatch only surfaces the first time a directive actually matters. Nothing guards it, either — `tests/test_deploy.py` reads the repo file, which is exactly the copy that is still correct. Diff the two when a restart follows a unit edit (#11 deploy). + +The copy is deliberate, for the same reason `/etc/replicator/.env` is not read from the repo: the live unit must survive a repo reset, a worktree switch, or a branch checkout that happens to be mid-edit. + +**Dev server workflow** (the `/health` app, port 8041 so a future live service stays up): + +```bash +set -a; . /etc/replicator/.env 2>/dev/null; . .env 2>/dev/null; set +a +uv run uvicorn src.api.main:app --host 0.0.0.0 --port 8041 --reload --log-config src/core/log_config.json +``` + +## Environment Variables + +In `.env` (dev/agent only — never read by the service): +- `GH_TOKEN` — GitHub PAT for this repo (used by `gh` CLI) +- `GH_TOKEN_ARCHIVER` / `GH_TOKEN_WATCHER` / `GH_TOKEN_CANNOBSERV` / `GH_TOKEN_SKILLS` — per-repo PATs. Cross-repo work is **filed as an issue**, never edited directly: each repo owns its own review, CI, and deploy cycle, and `main` is the deployed code. Pass the right one as `GH_TOKEN` for a given `gh` call. + +Read by neither env file — test-only, defined in `tests/conftest.py`: +- `REPLICATOR_TEST_REDIS_URL` — live broker for `@pytest.mark.integration`; default `redis://localhost:6379/15`. Must not resolve to db 0 (the fixture fails outright if it does) — see **Testing the bus** + +In `/etc/replicator/.env` (read by the service): +- `GOOGLE_APPLICATION_CREDENTIALS` — SA key for the wheelhouse mirror (`/etc/replicator/co-pypi-reader.json`) +- `REPLICATOR_REDIS_URL` — change-bus client URL; default `redis://localhost:6379/0` +- `REPLICATOR_BLOB_DIR` — temp-storage root for fetched bytes; default `blobs`. Resolved to an absolute path at store construction — `file://` URIs require it +- `REPLICATOR_BLOB_TTL_SECONDS` — how long a blob survives after it was **last referenced**; default `604800` (7 days). Measured from mtime, which the store refreshes on its short-circuit. The number is a published commitment to archiver (archiver#118), not a local tuning knob — raise it if a `content.blobs` consumer says it needs longer +- `REPLICATOR_BLOB_SWEEP_INTERVAL_SECONDS` — how often the tree is walked; default `900`. Also the staleness bound on the measured byte total the ceiling reads +- `REPLICATOR_BLOB_TEMP_GRACE_SECONDS` — how long a `.tmp` may live before the sweep treats it as debris; default `3600`. Deliberately unrelated to the TTL and far shorter — see **Retention** +- `REPLICATOR_BLOB_MAX_TOTAL_BYTES` — ceiling on everything the blob tree holds; default `2147483648` (2 GiB). Crossing it pauses fetching (`TransientFetchError`); it never shortens the TTL +- `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` — the most a command's own `timeout_seconds` may ask for; default `120`. Not a default (an omitted field still gets the driver's 30 s) but a ceiling, and a guard rather than a preference: the consume path is serial, so one issuer's timeout is a lien on every other command in the group. Over it ⇒ `PermanentFetchError`. Bounded above by the unit's `TimeoutStopSec` — change one, revisit the other +- `REPLICATOR_MAX_BLOB_BYTES` — ceiling on one fetched body; default `67108864` (64 MiB). A **storage** guard, not a memory one: co-core's fetch driver buffers the whole response before returning it, so the bytes are already resident when this is checked. Over the ceiling ⇒ `PermanentFetchError` ⇒ DLQ +- `REPLICATOR_CONSUMER_GROUP` — consumer group on `content.fetch`; default `replicator.fetch` +- `REPLICATOR_CONSUMER_NAME` — this worker's identity within the group; defaults to `replicator@`. Two workers must never share one — Redis tracks pending entries per consumer name, and a shared name makes independent `claim_stale` recovery impossible +- `REPLICATOR_CONSUMER_START_ID` — group start position; default `"$"` (new messages only), `"0"` drains the backlog. Applies **only at group creation** — once `replicator.fetch` exists, changing this also needs a manual `XGROUP SETID` +- `REPLICATOR_READ_BLOCK_MS` — blocking-read window; default `5000`. Bounds worst-case shutdown latency, so the unit's `TimeoutStopSec` must exceed it plus the handler budget **plus an in-flight sweep** — `asyncio.to_thread` puts the tree walk beyond cancellation, so SIGTERM waits it out +- `REPLICATOR_MIN_HOST_INTERVAL_SECONDS` — minimum spacing for a host with **no explicit policy**; default `1.0`. Since #19 this is the *fallback*, not the rule: the per-host numbers arrive on `content.fetch-policy` and an unknown, revoked, or not-yet-replayed host resolves here — never to unlimited, because a boot replay cannot tell a consumer whether the set it received is whole. 1.0 is Watcher's own `DEFAULT_MIN_INTERVAL`, chosen because it invents nothing. A wait ≤ `REPLICATOR_READ_BLOCK_MS` is slept through in the handler; a longer one raises `TransientFetchError` and parks the command, so the effective floor on a *parked* wait is `REPLICATOR_CLAIM_MIN_IDLE_MS`. **`0` no longer disables pacing outright** (#19 narrowed it): it is the fallback for unpublished hosts only, and a host with a policy is still paced by it — letting an env var veto a published value would invert the ownership split the charter settles. Capped at `3600`: past an hour the command parks and re-parks without ever dead-lettering (transient failures are exempt from the delivery ceiling) while the issuer's reaper concludes loss, so a fat-fingered extra zero should fail at startup rather than read as healthy. **It cannot be validated against what a producer might publish** — a published interval has no upper bound — so the strictness contract is enforced the only place it is knowable: a WARNING per host at apply time when a real policy turns out stricter than this number +- `REPLICATOR_CLAIM_MIN_IDLE_MS` — idle time before a pending entry may be reclaimed; default `60000`. Doubles as the retry cadence +- `REPLICATOR_MAX_DELIVERY_ATTEMPTS` — delivery ceiling for *unclassified* failures before DLQ; default `5`. Counted from XPENDING's delivery counter, which only advances on a reclaim ⇒ a bound in time, not retries +- `REPLICATOR_ERROR_BACKOFF_BASE_SECONDS` / `REPLICATOR_ERROR_BACKOFF_MAX_SECONDS` — backoff for a poll *cycle* that raised (broker outage); defaults `1.0` / `30.0`, escalating `base * 2**(n-1)` +- `REPLICATOR_MAX_CONSECUTIVE_CYCLE_FAILURES` — consecutive failed cycles before the worker exits so the unit restarts; default `20` (~8 min at the default backoff). Paired with the unit's `StartLimitIntervalSec=3600` / `StartLimitBurst=3` +- `REPLICATOR_DEDUPE_TTL_SECONDS` — lifetime of the `replicator:cmd:` dedupe key; default `86400` +- `REPLICATOR_LOG_LEVEL` — default `INFO`. Governs the **root** logger only, which is the whole tree for the worker. Under the dev server's `--log-config`, uvicorn's own `uvicorn` / `uvicorn.access` / `uvicorn.error` loggers are pinned `INFO` by `src/core/log_config.json` and do not follow it (nor did they under uvicorn's built-in config), so setting `WARNING` will not silence access lines; root itself is `INFO` from boot until the lifespan's `configure_logging()` applies this value +- `BUILD_ID` — git SHA stamped by the systemd unit's `ExecStartPre`; defaults to `"dev"` outside systemd diff --git a/docs/SKILLS.md b/docs/SKILLS.md index 1e1796a..0e5bfc3 100644 --- a/docs/SKILLS.md +++ b/docs/SKILLS.md @@ -41,6 +41,7 @@ Every override's `SKILL.md` must declare `overrides:` and `override-reason:` in | Skill | Purpose | |---|---| +| `curating-context` | Curates `AGENTS.md` and its live reference docs against a token budget | | `enforcing-architecture` | Turns an accepted architecture finding into an executable fitness function | | `init-project-fastapi` | Bootstraps a CannObserv FastAPI service (this repo's own foundation) | | `init-socraticode` | Installs and indexes SocratiCode semantic code search | @@ -60,6 +61,13 @@ is accepted with a `N: fix + fitness` or bare `N: fitness` directive — without directive fails to resolve. The daily auto-refresh hook bumps the submodule pointer but never creates per-skill symlinks, so linking a newly published skill stays a manual step (#13). +**`curating-context` is pinned at v1.2 until the wave-B comparison resolves (#22).** The twelve +cohort repos are the held-out validation split for the skill itself: a proposed change is tried on +one arm and scored against the other, and Replicator's first curation is this arm's data point. +Bumping the vendored pointer past v1.2 before that resolves puts two skill versions inside one arm +and `score-cohort.sh` returns INCONCLUSIVE rather than a verdict. The daily auto-refresh hook bumps +the submodule pointer, so this pin is a review obligation, not something the tooling enforces. + ### From `obra-superpowers` | Skill | Purpose | diff --git a/docs/STORAGE.md b/docs/STORAGE.md new file mode 100644 index 0000000..60a1b88 --- /dev/null +++ b/docs/STORAGE.md @@ -0,0 +1,22 @@ +# Replicator Temp Storage + +The blob tree: how a fingerprint becomes a path, what the three populations under +`REPLICATOR_BLOB_DIR` are, and why the sweep reaps some of them and never others. +Replicator is the producer in archiver's temp-cache protocol, so the cleanup +obligation is Replicator's. + +## The blob tree + +- **A blob is `file://///.bin`.** Sharded two levels to bound directory fan-out; the extension is a constant `.bin`, **never** derived from `media_type` — identical octets can arrive under different Content-Types, and two paths for one fingerprint would defeat `exists()` as a short-circuit. Writes go through a temp file + `os.replace`: presence at a content-addressed path is what readers take as proof the bytes are complete. Design: `docs/plans/2026-07-31-replicator-mvp-open-questions-design.md`. +- **Blob modes are set on creation only.** Files land at `0644`, directories the worker creates at `0755` — both by explicit `chmod`, since `mkstemp` creates at `0600` and `mkdir`'s mode is masked by the umask. A directory that **already exists is never re-chmod'd**: it belongs to whoever provisioned it, and `chmod` on an unowned-but-writable mount raises `EPERM`. The cost is a silent trap — a `0700` level anywhere in the chain stores and publishes normally while no other service can open the `blob_uri` — so `warn_if_unreachable` walks `blob_dir` **and every parent** at startup and names each blocking level. Traversal needs `+x` all the way up, and the likeliest mistake is a restrictive parent over a fine leaf. `src/storage/local.py::ensure_directory`, `src/worker/main.py::warn_if_unreachable`. + +## Retention + +`docs/plans/2026-07-31-replicator-mvp-open-questions-design.md` §4 scope-cut retention; #5 settles it. Replicator is the producer in archiver's temp-cache protocol, where **the producer cleans up**. + +- **The TTL runs from last reference, not first store.** `store` short-circuits on an existing content-addressed path but its caller publishes a fresh `blob_available` either way, so a re-fetch of unchanged bytes would otherwise announce a blob already partway through its TTL. `LocalBlobStore` therefore `os.utime`s on the short-circuit branch, swallowing `ENOENT` — the sweep can unlink between the existence check and the touch, and the fallout of that race is a `blob_uri` that fails to open, not a dead-lettered command. +- **The blob tree holds three populations, and they are not interchangeable.** Finished blobs (`//.bin`) reap on the TTL; in-flight temporaries (`...tmp`) are **not garbage** — reaping one makes the writer's `os.replace` fail with `ENOENT` and dead-letters a good command — so the sweep matches `*.bin`, never `iterdir()`, and ages temps out on their own much longer grace. Empty shard directories go **last**, by `rmdir` only, whose refusal to touch a non-empty directory is the safety property. +- **The ceiling is backpressure, not a faster clock.** Over `REPLICATOR_BLOB_MAX_TOTAL_BYTES` the byte path raises `TransientFetchError` *before* fetching, so the command stays in the PEL and returns via `claim_stale` once a sweep frees space. What it measures is everything the tree holds — surviving blobs **and** the temporaries the sweep is waiting out, since a crash loop fills the disk with debris the ceiling would otherwise not see. The per-population counts stay split in the sweep log so a rising temp count cannot hide inside a healthy blob one. Reaping a blob still inside its TTL to make room would convert a local disk problem into a `blob_uri` another repo cannot open — the one failure mode with no local symptom. +- **One `BlobUsage`, two writers.** The sweep's `observe` is the measured total; the byte path's `add` is the estimate between sweeps, because a burst can cross the ceiling long before the tree is walked again. Wiring the two halves to separate instances leaves both individually correct and the guard permanently unreachable — `tests/worker/test_main.py` pins the identity. +- **Orphans are recorded where they are exact.** A publish that fails after the store leaves bytes with no fact and no `command_id`, invisible to any query starting from `content.blobs`. `src/worker/handler.py::_publish` logs the fingerprint at that moment and re-raises untouched; the sweep then treats orphans as ordinary aged blobs. Reconciling the tree against `content.blobs` instead would make a *delete* decision depend on another service's stream-trimming policy. +- **The sweep runs in the worker, not a systemd timer.** A timer survives a crashed worker, but a worker that is not running is not writing blobs either. It rides the same stop event as the consume loop (`src/worker/loop.py::park`) and walks the tree via `asyncio.to_thread`, so retention never becomes a source of consume-path latency. A failed sweep is absorbed and retried next cycle — retention is not load-bearing for correctness, and the ceiling is the guard for a tree that cannot be reaped. diff --git a/docs/STYLE.md b/docs/STYLE.md new file mode 100644 index 0000000..6f2185a --- /dev/null +++ b/docs/STYLE.md @@ -0,0 +1,13 @@ +# Replicator Style & Logging + +Code style beyond the conventions `AGENTS.md` keeps inline. Today that is the +logging stack: one formatter, two installers, and the two places its output is +deliberately not JSON. + +## Logging + +**One formatter, two installers.** `build_json_formatter()` is the single definition of the JSON schema (`timestamp`, `level`, `logger`, `message`). `configure_logging()` installs it on the root logger; `src/core/log_config.json` names the *same factory* through dictConfig's `"()"` key, so there is no second fmt string to drift. The dev server must be launched with `--log-config src/core/log_config.json` — uvicorn's `uvicorn` / `uvicorn.access` / `uvicorn.error` loggers ship with `propagate=False` and their own plain-text handlers, so a root-only config never reaches them and the output is half JSON, half plain text. **The worker runs no uvicorn**, so `replicator.service` needs no `--log-config`; its `ExecStart` is `python -m src.worker.main` and `configure_logging()` is the whole story there (#14). + +**Not everything in the journal is JSON.** The claim above is about the *app's own records*. `replicator.service`'s wheelhouse-sync `ExecStartPre` writes a **plain-text** line to journald on every start — `wheelhouse in sync: N downloaded, M already present -> …` on the happy path, `error: could not sync gs://…` on the non-fatal failure path (stderr, which journald captures the same way) — and that is by design, not a gap: it runs `uv run --no-project` — before the deploy's `uv sync`, in an environment holding `google-cloud-storage` and nothing else — so it cannot import `build_json_formatter()` and making it emit JSON would mean a hand-maintained second copy of the schema in the one file that structurally cannot single-source it. A shipper reading journald natively is unaffected (the message is a field alongside `_SYSTEMD_UNIT` / `_PID`); a pipeline that `json.loads` every `MESSAGE` must tolerate these two lines — the failure one especially, since it appears exactly when something is already wrong (#15, skills#83). + +**The colour strip is a filter on the loggers, deliberately.** uvicorn attaches an ANSI-coloured duplicate of each lifecycle message as `extra={"color_message": ...}`, and every extra reaches the JSON payload. `ColorMessageFilter` deletes it from the record at its source — not via the formatter's `reserved_attrs`, and not on the stdout handler. Both alternatives scope the fix to *this* sink: a handler that builds its payload from the record's `__dict__` rather than a `logging.Formatter` resurrects the field, and OpenTelemetry's `LoggingHandler` is exactly that (its own reserved list does not cover `color_message`). Mutating the record once means the strip survives a sink swap with no failing test to warn you it had stopped working. `tests/core/test_logging.py` pins the filter's placement, not just its effect. diff --git a/docs/TESTING.md b/docs/TESTING.md new file mode 100644 index 0000000..a4b252f --- /dev/null +++ b/docs/TESTING.md @@ -0,0 +1,16 @@ +# Testing Replicator + +Test layout is mirrored from `src/`, and the suite runs against fakeredis by +default. What follows is the part that is not derivable from reading the tests: +where the fake diverges from the real broker, and which keys a live-broker run is +allowed to touch. + +## Testing the bus + +**Testing the bus.** `tests/conftest.py` ships a `fake_redis` fixture (fakeredis, Streams-capable) — consumer-group behaviour is testable without a broker, and assertions should read the broker's own view (`xinfo_groups` / `xinfo_consumers`) rather than co-core's private attributes, which are not a stable contract. Anything that genuinely needs the live Archiver-operated Redis goes behind `@pytest.mark.integration` and is excluded by default. + +**Where fakeredis diverges.** It is sound for consumer-group *mechanics* — what state a command leaves behind — but diverges on *lifecycle* and *blocking* semantics: it registers a consumer on an empty `XREADGROUP` (real Redis waits for a delivery, GH #3) and it ignores `block` (worked around by `IDLE_SLEEP_SECONDS` in `src/worker/loop.py`). Rule of thumb: an assertion about **what state results** is safe against the fake; an assertion about **when Redis does something** needs a live broker. Both divergences were found by running against the real server, not by the suite. + +Live-broker tests use the `real_redis` fixture (`tests/conftest.py`), which connects to `REPLICATOR_TEST_REDIS_URL` (default `redis://localhost:6379/15`), skips when nothing answers (an *auth* failure re-raises — a misconfigured broker must not pass as a skip), expires stray `replicator.itest.*` keys from crashed runs once per session, and refuses db 0 outright — db 0 carries the live `content.fetch` stream that the running `replicator.service` consumes, so a test frame written there would be fetched for real. Confine such tests to scratch stream keys via the `scratch_topic` fixture (`tests/worker/conftest.py`), whose teardown also removes `.dlq`; the database guard is the backstop, not the plan. + +**One namespace the sweeper cannot reach.** `process_message` writes `replicator:cmd:`, a constant prefix outside `replicator.itest.*`, so an end-to-end test deletes its own keys via the `dedupe_keys` fixture and shortens their TTL. `test_an_end_to_end_run_only_creates_predictable_keys` asserts the whole promise: every key a run creates is either an itest stream or a dedupe key. From 0f4fded9800592f71de5c3b777e1a6eabb408adb Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Thu, 6 Aug 2026 06:15:16 +0000 Subject: [PATCH 3/4] #22 chore: suspend the daily skills refresh for the v1.2 hold MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `skills-submodule-update.sh` runs `git submodule update --remote --merge -- skills-vendor/` once per UTC day on main and commits the result, which would move `gregoryfoster-skills` past `3fc7b71` within a day and put two skill versions inside the wave-A arm. The hook takes no per-submodule exclusion and one submodule carries every `gregoryfoster` skill, so pinning `curating-context` alone is not expressible. Removing the SessionStart entry is the only remedy a consumer repo has today — the same one `CannObserv/cli` took (gregoryfoster/skills#100). The hook script stays in place, so restoring the refresh is a one-entry edit. `docs/SKILLS.md` records the pin, why the suspension is blunter than the problem, the manual update path while it holds, and when to restore. Co-Authored-By: Claude Opus 5 (1M context) --- .claude/settings.json | 8 -------- docs/SKILLS.md | 32 ++++++++++++++++++++++++++------ 2 files changed, 26 insertions(+), 14 deletions(-) diff --git a/.claude/settings.json b/.claude/settings.json index aa2f60c..2e55e23 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -1,14 +1,6 @@ { "hooks": { "SessionStart": [ - { - "hooks": [ - { - "type": "command", - "command": "bash .claude/hooks/skills-submodule-update.sh" - } - ] - }, { "hooks": [ { diff --git a/docs/SKILLS.md b/docs/SKILLS.md index 0e5bfc3..5f332a0 100644 --- a/docs/SKILLS.md +++ b/docs/SKILLS.md @@ -61,12 +61,32 @@ is accepted with a `N: fix + fitness` or bare `N: fitness` directive — without directive fails to resolve. The daily auto-refresh hook bumps the submodule pointer but never creates per-skill symlinks, so linking a newly published skill stays a manual step (#13). -**`curating-context` is pinned at v1.2 until the wave-B comparison resolves (#22).** The twelve -cohort repos are the held-out validation split for the skill itself: a proposed change is tried on -one arm and scored against the other, and Replicator's first curation is this arm's data point. -Bumping the vendored pointer past v1.2 before that resolves puts two skill versions inside one arm -and `score-cohort.sh` returns INCONCLUSIVE rather than a verdict. The daily auto-refresh hook bumps -the submodule pointer, so this pin is a review obligation, not something the tooling enforces. +**`curating-context` is pinned at v1.2 (`3fc7b71`) until the wave-B comparison resolves (#22).** The +twelve cohort repos are the held-out validation split for the skill itself: a proposed change is +tried on one arm and scored against the other, and Replicator's first curation is this arm's data +point. Bumping the vendored pointer past v1.2 before that resolves puts two skill versions inside one +arm and `score-cohort.sh` returns INCONCLUSIVE rather than a verdict. + +**So the daily auto-refresh is suspended for the duration of the hold.** The hook's `SessionStart` +entry is removed from `.claude/settings.json`; `.claude/hooks/skills-submodule-update.sh` itself is +left in place, so restoring the refresh is a one-entry edit and not a re-install. Suspension is +blunter than the problem — it also stops the `obra-superpowers` refresh and the `.skills/doctor.sh` +self-heal commit, neither of which has anything to do with the hold — but it is the only remedy a +consumer repo currently has: `git submodule update --remote --merge -- skills-vendor/` takes no +per-submodule exclusion, and one submodule carries every `gregoryfoster` skill, so pinning +`curating-context` alone is not expressible. Tracked upstream as +[gregoryfoster/skills#100](https://github.com/gregoryfoster/skills/issues/100), which proposes a +committed pin file the hook consults; restore the `SessionStart` entry when that lands or when the +hold ends, whichever comes first. + +While the refresh is suspended, updating vendor skills is manual: + +```bash +git -C skills-vendor/obra-superpowers fetch origin && git -C skills-vendor/obra-superpowers checkout origin/HEAD +bash .skills/doctor.sh +``` + +Leave `skills-vendor/gregoryfoster-skills` at `3fc7b71` until the hold ends. ### From `obra-superpowers` From a60a9b1c707f5a7495232889357a05b58bb0da57 Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Thu, 6 Aug 2026 06:34:11 +0000 Subject: [PATCH 4/4] #22 fix: apply first-round code review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. tests/core/test_logging.py — docs/DEPLOYMENT.md is a fourth uvicorn command site since the Server Lifecycle demotion; it was outside UVICORN_COMMAND_SOURCES, so the --log-config guard passed by not looking. 2. README.md — pointed at AGENTS.md for "Not everything in the journal is JSON", which moved to docs/STYLE.md on this branch. 3. docs/DEPLOYMENT.md — two relocated bullets end "see **Testing the bus**" and "see **Retention**", intra-file references at the branch point and cross-file now. Destinations named on continuation lines so the original lines stay verbatim. 4. docs/ARCHITECTURE.md — the layout tree read "docs/ — Reference docs (COMMANDS, SKILLS)". Warrant: disproven. `ls docs/*.md` returns seven, and this branch is what made it false. The roster now points at the Detail Docs index rather than re-listing. 5. .socraticodecontextartifacts.json — `architecture` named ./README.md while ./docs/ARCHITECTURE.md was `architecture-detail`, implying a parent/child pair that does not exist. Now `readme` and `architecture`. Re-embedded via codebase_context_index. 6. docs/ARCHITECTURE.md — dropped the duplicated "Replicator is a consumer first" lead-in; the line stays inline in AGENTS.md. 7. docs/SKILLS.md — records skills#99: the write-guard hook symlink is outside doctor.sh's heal scope and dangles on a submodule-less checkout, failing on every edit. Also completes the manual-update snippet, which omitted the commit the suspended hook used to perform. 8. README.md — its 24-row environment table duplicated docs/DEPLOYMENT.md with neither marked authoritative. Now a five-row table of what this VM actually overrides, pointing at DEPLOYMENT.md for the rest. prove-no-loss.sh reports exactly one line dropped — finding 4, with the warrant above. Everything else is still inline or relocated verbatim. Co-Authored-By: Claude Opus 5 (1M context) --- .socraticodecontextartifacts.json | 6 ++--- README.md | 38 +++++++++++-------------------- docs/ARCHITECTURE.md | 4 ++-- docs/DEPLOYMENT.md | 2 ++ docs/SKILLS.md | 16 +++++++++++++ tests/core/test_logging.py | 1 + 6 files changed, 37 insertions(+), 30 deletions(-) diff --git a/.socraticodecontextartifacts.json b/.socraticodecontextartifacts.json index e2639a9..2cb0b8a 100644 --- a/.socraticodecontextartifacts.json +++ b/.socraticodecontextartifacts.json @@ -6,9 +6,9 @@ "description": "Authoritative project guide for agents and contributors, curated to a 6,000-token budget: it carries the rules needed on nearly every task and routes to the reference docs for everything else (see its Detail Docs index). Inline: the worker-first shape (the bus consumer is the primary process; FastAPI /health is dev-only), the co-core wheelhouse mechanism (gs://co-gcs-pypi via [tool.uv] find-links, not PyPI) and the load-bearing patch floor, the SocratiCode code-exploration policy, the hard boundary between /etc/replicator/.env and the repo .env, the Redis >=7.0 floor and Archiver's ownership of the broker, the nine bus rules that bind every change (at-least-once idempotency with two keys, store-then-publish, read(count=1), fail-loud from_wire with a global dispatch table, DLQ vs retry, no outbox on the consume path, the three stream kinds, seed_fetch.py as the only writer to content.fetch, the two normative contracts), the common commands, and the commit/logging/date conventions. Module layout, the reasoning behind each bus contract, blob-tree and retention semantics, deployment topology and the full environment-variable reference, fakeredis divergences, and the logging stack live in docs/ARCHITECTURE.md, docs/STORAGE.md, docs/DEPLOYMENT.md, docs/TESTING.md, and docs/STYLE.md respectively." }, { - "name": "architecture", + "name": "readme", "path": "./README.md", - "description": "Top-level README. What Replicator is within the Cannabis Observer cluster: the command -> fetch -> fingerprint -> temp-store -> fact loop, why the worker rather than the API is the deployed process, setup via the wheelhouse sync, the service-visible environment variable table, and the systemd deploy recipe." + "description": "Top-level README. What Replicator is within the Cannabis Observer cluster: the command -> fetch -> fingerprint -> temp-store -> fact loop, content.fetch-policy as the third stream kind, why the worker rather than the API is the deployed process, setup via the wheelhouse sync, the two-env-file boundary with a table of only what this VM overrides (docs/DEPLOYMENT.md is the authoritative variable reference), the seed_fetch.py smoke test, the dev server and why it needs --log-config, and the systemd deploy recipe." }, { "name": "design-plans", @@ -26,7 +26,7 @@ "description": "Normative charter for what Replicator is allowed to become — sibling to the issuer contract, which settles the wire while this settles the service. The rule: Replicator owns the mechanics of acquiring bytes from a network and holding them briefly, never why, when, or what they mean. Three tests for any proposed capability, field, or setting (durable per-resource history -> issuer; cross-command coordination over a resource only the fetcher can see -> Replicator; expressible without domain vocabulary or it is the issuer's), resolved when both fire by mechanism-to-Replicator / policy-to-the-issuer / config-over-the-bus. Worked examples: per-host politeness (the hard case, where the tests do not resolve to one side) and conditional GET (ETag/Last-Modified are durable per-resource history, so the fact returns them and the next command replays them as headers — conditional GET with zero per-URL state here). Config taxonomy of three channels (env for host facts, command for the occasion, a policy stream for cluster policy needing cross-command state) with a fourth — an inbound admin HTTP API — rejected by name; ingress is read-only liveness and self-description, and the invariant that matters is asserted about the worker, which binds no port, not the dev-only FastAPI app. The policy stream SHIPPED in #19 on co-core v0.7.7 (cannobserv#285) and is named **content.fetch-policy with a hyphen** — the dotted content.fetch.policy this document originally proposed collides with the .dlq derivation of the command stream. Last-write-wins per host, replayed from 0-0 at boot via the groupless AsyncBusTailReader and tailed thereafter, MAXLEN-trimmed by the producer because periodic full republication onto an untrimmed stream is unbounded growth, a Redis hash rejected for having no schema_version and coupling to a key name. Documents the consumer-side rules that fail silently if missed: an unknown OR revoked host resolves to the conservative env default and never to unlimited (revoked is a tombstone meaning 'no explicit policy', so branch on revoked before reading min_interval_seconds, which is None there); 0.0 is a legal interval and is falsy; the default's strictness cannot be asserted at startup because a published interval has no upper bound, so it is reported per host at apply time instead; arrival order is not publication order, so the map guards on occurred_at per host; and from_wire's dispatch table is global, so a foreign payload decodes cleanly into the wrong model and needs an isinstance check. Enforcement idiom: park in the PEL, return via claim_stale — whose 60s granularity misses Watcher's 1.0s baseline by 60x, resolved by splitting the wait by duration (sleep under one poll window, park above it). REPLICATOR_MIN_HOST_INTERVAL_SECONDS is now the FALLBACK rather than the rule, and 0 no longer disables pacing outright. Two known limitations recorded rather than omitted: blob_uri's host-local file:// coupling (#7), and the host asked for not being the host reached under redirects, which #19 did NOT resolve. Enforced by tests/test_boundaries.py in CI: eight invariants including an AST scan of src/ for domain nouns in identifiers and string literals, a recursive route walk, no locally-defined wire models (payload shapes, not producer-owned token vocabularies like FailureReason), and REPLICATOR_-prefixed settings with BUILD_ID exempted by name. Written for #12, updated for #19." }, { - "name": "architecture-detail", + "name": "architecture", "path": "./docs/ARCHITECTURE.md", "description": "Module-by-module layout of src/ (worker, storage, api, core) with the job each file owns, and the full reasoning behind every bus contract Replicator consumes or publishes: the issuer contract and boundaries charter pointers, content.blobs carrying both blob_available and fetch_failed, the six optional passthrough fields on blob_available and why None means nobody said it, the FailureReporter seam, per-command headers/timeout_seconds validation and why unsendable options are refused rather than clamped, the pacing split between sleeping and parking, the groupless content.fetch-policy reader and the four ways to apply a policy message wrongly, at-least-once idempotency, claim_stale as the retry path, XPENDING delivery accounting, and the message-vs-cycle failure distinction. Read this before changing what any stream carries." }, diff --git a/README.md b/README.md index 691f015..699cfca 100644 --- a/README.md +++ b/README.md @@ -67,10 +67,10 @@ For local work, load both: set -a; . /etc/replicator/.env 2>/dev/null; . .env 2>/dev/null; set +a ``` -Variables the service uses (all in `/etc/replicator/.env`). The **Default** column is the -value baked into `src/core/config.py` — several are overridden on the VM, so check -`/etc/replicator/.env` before assuming a default applies. On this deployment -`REPLICATOR_BLOB_DIR` is `/var/lib/replicator/blobs`, not the `blobs` shown below. +Variables the service uses all live in `/etc/replicator/.env`. **[docs/DEPLOYMENT.md](docs/DEPLOYMENT.md) +is the authoritative reference** — every variable, its default, and the reasoning behind +each one. Below is only what this deployment actually overrides; anything absent from +`/etc/replicator/.env` runs on the default baked into `src/core/config.py`. **If you pre-create the blob directory, make it and every parent traversable** (`0755`). The worker sets modes only on directories it creates itself — an existing one keeps @@ -86,30 +86,18 @@ unchanged bytes restarts it. Disk pressure never shortens it: over `REPLICATOR_BLOB_MAX_TOTAL_BYTES` the worker stops fetching and leaves commands on the bus rather than deleting bytes a consumer was promised. -| Variable | Default | Purpose | +| Variable | Set on this VM | Purpose | |---|---|---| -| `GOOGLE_APPLICATION_CREDENTIALS` | — | SA key for the wheelhouse mirror (`/etc/replicator/co-pypi-reader.json`) | +| `GOOGLE_APPLICATION_CREDENTIALS` | `/etc/replicator/co-pypi-reader.json` | SA key for the wheelhouse mirror | | `REPLICATOR_REDIS_URL` | `redis://localhost:6379/0` | Change-bus client URL | -| `REPLICATOR_BLOB_DIR` | `blobs` | Temp-storage root for fetched bytes. Resolved to an absolute path — `file://` URIs require it | -| `REPLICATOR_BLOB_TTL_SECONDS` | `604800` | How long a blob survives after it was **last referenced** (7 days). A commitment made to archiver (archiver#118), not a local knob — raise it if a `content.blobs` consumer needs longer | -| `REPLICATOR_BLOB_SWEEP_INTERVAL_SECONDS` | `900` | How often the tree is walked. Also the staleness bound on the size the ceiling reads | -| `REPLICATOR_BLOB_TEMP_GRACE_SECONDS` | `3600` | How long a `.tmp` may live before the sweep treats it as debris. Far shorter than the TTL — a temporary exists only across a single write | -| `REPLICATOR_BLOB_MAX_TOTAL_BYTES` | `2147483648` | Ceiling on everything the tree holds (2 GiB). Crossing it **pauses fetching**; it never reaps a blob still inside its TTL | -| `REPLICATOR_MAX_BLOB_BYTES` | `67108864` | Ceiling on one fetched body (64 MiB). A *storage* guard, not a memory one — co-core's fetch driver buffers the whole response first. Over it ⇒ DLQ | -| `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` | `120` | The most a command's own `timeout_seconds` may ask for (#11). Not a default — an omitted field still gets the driver's 30 s — but a ceiling, and a guard rather than a preference: the consume path is serial, so one issuer's timeout is a lien on every other command in the group. Over it ⇒ DLQ. Bounded above by the unit's `TimeoutStopSec` | -| `REPLICATOR_CONSUMER_GROUP` | `replicator.fetch` | Consumer group on `content.fetch` | +| `REPLICATOR_BLOB_DIR` | `/var/lib/replicator/blobs` | Temp-storage root — **not** the `blobs` default | | `REPLICATOR_CONSUMER_NAME` | `replicator@` | This worker's identity in the group — never share one | -| `REPLICATOR_CONSUMER_START_ID` | `$` | Group start position. Applies only at group *creation*; changing it later also needs `XGROUP SETID` | -| `REPLICATOR_READ_BLOCK_MS` | `5000` | Blocking-read window, used by both the consume loop and the policy tail (#19). Bounds shutdown latency, so the unit's `TimeoutStopSec` must exceed it plus `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` (the handler's worst-case budget since #11) and an in-flight sweep — `tests/test_deploy.py` pins the first two terms. The two readers block concurrently, so this contributes one term rather than two | -| `REPLICATOR_MIN_HOST_INTERVAL_SECONDS` | `1.0` | Minimum spacing for a host with **no explicit policy** (#12, #19). Since #19 the per-host numbers arrive on `content.fetch-policy`; this is the fallback an unknown, revoked, or not-yet-replayed host resolves to — never "unlimited". Matches Watcher's own `DEFAULT_MIN_INTERVAL`. A wait under `REPLICATOR_READ_BLOCK_MS` is slept through, a longer one parks the command for the next reclaim. **`0` does not disable pacing** — it is the fallback for unpublished hosts only, and a host with a policy is still paced by it. Capped at `3600` | -| `REPLICATOR_CLAIM_MIN_IDLE_MS` | `60000` | Idle time before a pending entry may be reclaimed — also the retry cadence | -| `REPLICATOR_MAX_DELIVERY_ATTEMPTS` | `5` | Deliveries of an *unclassified* failure before the DLQ | -| `REPLICATOR_DEDUPE_TTL_SECONDS` | `86400` | Lifetime of the `replicator:cmd:` dedupe key | -| `REPLICATOR_ERROR_BACKOFF_BASE_SECONDS` | `1.0` | Backoff after a failed poll cycle (broker outage), escalating `base * 2**(n-1)` | -| `REPLICATOR_ERROR_BACKOFF_MAX_SECONDS` | `30.0` | Cap on that backoff | -| `REPLICATOR_MAX_CONSECUTIVE_CYCLE_FAILURES` | `20` | Failed cycles before the worker exits so the unit restarts (~8 min). Paired with the unit's `StartLimitIntervalSec` | | `REPLICATOR_LOG_LEVEL` | `INFO` | Root log level | -| `BUILD_ID` | `dev` | Git SHA, stamped by the unit's `ExecStartPre` | + +`BUILD_ID` is stamped by the unit's `ExecStartPre` rather than set in the env file. Every +other `REPLICATOR_*` setting — the blob TTL and ceilings, the consumer group and start id, +the read window and pacing fallback, the reclaim and backoff numbers — is on its default; +see [docs/DEPLOYMENT.md](docs/DEPLOYMENT.md) for what each one is and why. ## Seeding a fetch @@ -174,7 +162,7 @@ Production secrets live in `/etc/replicator/.env` (managed manually on the VM, n The unit's `ExecStartPre` writes the current git SHA to `/run/replicator/build-id` and exposes it as `BUILD_ID`, asserts the Redis `>=7.0` floor via `scripts/check_redis_floor.sh`, and refreshes the wheelhouse via `scripts/sync_wheelhouse.py` — whose journald output is **plain text, not JSON**, by -design (see [AGENTS.md](AGENTS.md), "Not everything in the journal is JSON"). +design (see [docs/STYLE.md](docs/STYLE.md), "Not everything in the journal is JSON"). Because `ExecStart` runs `--frozen --no-sync`, run `uv sync --frozen` as part of the deploy, before `systemctl restart`. diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index c49100f..a4ca0f2 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -30,7 +30,7 @@ src/core/config.py — Settings / env access (see Environment Variables) scripts/ — sync_wheelhouse.py, check_redis_floor.sh, seed_fetch.py scripts/seed_fetch.py — the MVP command issuer; publishes content.fetch, --watch tails the facts tests/ — Mirrors src/ structure; integration tests in `@pytest.mark.integration` -docs/ — Reference docs (COMMANDS, SKILLS) +docs/ — Reference docs; the Detail Docs index in AGENTS.md is the roster docs/contracts/ — Normative contracts, linked to from sibling repos: the issuer-facing half of the wire, and the boundaries charter (what Replicator may become) docs/plans/ — Implementation plans deploy/ — Systemd unit + deployment config @@ -39,7 +39,7 @@ deploy/ — Systemd unit + deployment config ## Bus Conventions -Replicator is a **consumer** first. Follow the conventions co-core and the archiver producer established: +Each rule `AGENTS.md` states in one line, with the reasoning that makes it non-negotiable: - **The issuer contract is written down and lives here.** `docs/contracts/content-fetch-issuer-contract.md` is the normative statement of what a `content.fetch` producer must do — per-occasion `command_id`, `url` is not a correlation key, persist the `command_id → domain` map before publishing, correlate idempotently, and keep a reaper as a backstop for the outcomes no fact can carry. Issuer-side repos (Watcher, Phase 4) link to it rather than copying it. Anything asserted there is asserted about this repo's code; change one, change both (#8). - **What Replicator is allowed to become is also written down.** `docs/contracts/replicator-boundaries.md` is the sibling charter: mechanism to Replicator, policy to the issuer, config over the bus, and an inbound admin HTTP API rejected by name. Run its three tests against any proposed capability, field, or setting before writing code — a database, domain vocabulary, or a write route is reached one defensible step at a time, not in one commit. `tests/test_boundaries.py` enforces eight invariants in CI, including the one that catches the regression review misses: an AST scan of `src/` for domain nouns in identifiers *and* string literals. Known violation recorded and pinned rather than omitted — `blob_uri` is host-local `file://` (#7). Change the charter and the tests together (#12). diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 15bbe80..ca4ed90 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -56,6 +56,7 @@ In `.env` (dev/agent only — never read by the service): Read by neither env file — test-only, defined in `tests/conftest.py`: - `REPLICATOR_TEST_REDIS_URL` — live broker for `@pytest.mark.integration`; default `redis://localhost:6379/15`. Must not resolve to db 0 (the fixture fails outright if it does) — see **Testing the bus** + in [TESTING.md](TESTING.md) In `/etc/replicator/.env` (read by the service): - `GOOGLE_APPLICATION_CREDENTIALS` — SA key for the wheelhouse mirror (`/etc/replicator/co-pypi-reader.json`) @@ -64,6 +65,7 @@ In `/etc/replicator/.env` (read by the service): - `REPLICATOR_BLOB_TTL_SECONDS` — how long a blob survives after it was **last referenced**; default `604800` (7 days). Measured from mtime, which the store refreshes on its short-circuit. The number is a published commitment to archiver (archiver#118), not a local tuning knob — raise it if a `content.blobs` consumer says it needs longer - `REPLICATOR_BLOB_SWEEP_INTERVAL_SECONDS` — how often the tree is walked; default `900`. Also the staleness bound on the measured byte total the ceiling reads - `REPLICATOR_BLOB_TEMP_GRACE_SECONDS` — how long a `.tmp` may live before the sweep treats it as debris; default `3600`. Deliberately unrelated to the TTL and far shorter — see **Retention** + in [STORAGE.md](STORAGE.md) - `REPLICATOR_BLOB_MAX_TOTAL_BYTES` — ceiling on everything the blob tree holds; default `2147483648` (2 GiB). Crossing it pauses fetching (`TransientFetchError`); it never shortens the TTL - `REPLICATOR_MAX_FETCH_TIMEOUT_SECONDS` — the most a command's own `timeout_seconds` may ask for; default `120`. Not a default (an omitted field still gets the driver's 30 s) but a ceiling, and a guard rather than a preference: the consume path is serial, so one issuer's timeout is a lien on every other command in the group. Over it ⇒ `PermanentFetchError`. Bounded above by the unit's `TimeoutStopSec` — change one, revisit the other - `REPLICATOR_MAX_BLOB_BYTES` — ceiling on one fetched body; default `67108864` (64 MiB). A **storage** guard, not a memory one: co-core's fetch driver buffers the whole response before returning it, so the bytes are already resident when this is checked. Over the ceiling ⇒ `PermanentFetchError` ⇒ DLQ diff --git a/docs/SKILLS.md b/docs/SKILLS.md index 5f332a0..eb1b87d 100644 --- a/docs/SKILLS.md +++ b/docs/SKILLS.md @@ -84,10 +84,26 @@ While the refresh is suspended, updating vendor skills is manual: ```bash git -C skills-vendor/obra-superpowers fetch origin && git -C skills-vendor/obra-superpowers checkout origin/HEAD bash .skills/doctor.sh +git add skills-vendor/obra-superpowers && git commit -m "chore: update skills submodules" ``` +The commit is the step the suspended hook used to perform; without it the new pointer is +discarded by the next checkout. + Leave `skills-vendor/gregoryfoster-skills` at `3fc7b71` until the hold ends. +## The write-guard hook dangles on a submodule-less checkout + +`curating-context` installs `.claude/hooks/context-budget-guard.sh` as a symlink into the +vendored skill, but `doctor.sh` scans `skills/*` only — `.claude/hooks/*` is outside its heal +scope. On a checkout where the submodule is not initialized (fresh clone, `git worktree add`, +shallow CI clone) the hook path dangles and the wired `PostToolUse` command fails with +`No such file or directory` on **every** `Edit`/`Write`/`MultiEdit`, naming a path that `ls` +shows as present. Run `bash .skills/doctor.sh` — it initializes the submodule and so heals the +hook as a side effect, even though it never inspects it. Tracked upstream as +[gregoryfoster/skills#99](https://github.com/gregoryfoster/skills/issues/99); the guard itself +is correctly non-blocking once it resolves. + ### From `obra-superpowers` | Skill | Purpose | diff --git a/tests/core/test_logging.py b/tests/core/test_logging.py index 5ce26c8..155592f 100644 --- a/tests/core/test_logging.py +++ b/tests/core/test_logging.py @@ -26,6 +26,7 @@ "README.md", "AGENTS.md", "docs/COMMANDS.md", + "docs/DEPLOYMENT.md", "deploy/replicator.service", ) UVICORN_INVOCATION = "uvicorn src.api.main:app"