Repository navigation
Become the cohort text-extraction processor: consume content.process, extract from one source_spec, store canonical text by hash, publish content.derived #1
Description
Recreated 2026-09-29 from CannObserv/observo#629. Observo is private, and GitHub refuses a private → public transfer. Observo is no longer the processor: this issue tracks Processor v1, the cohort's headless discrete-transform service, on VM
co-processor. The founding spec,docs/specs/2026-09-29-processor-service-design.md, amends the requirements below:
- output goes to Processor's own bucket,
gs://co-gcs-processor/blobs/<sha256>.bin(req. 5–6)- the ACL user and group are
processor/processor.process(req. 1, ACL userprocessorfor the content.process consumer; retireobservo's grants broker#75)- co-core is pinned
==0.19.4to match Watcher (req. 4)- isolation comes from Processor's own VM (req. 7)
The original body and its three comments follow verbatim. Bare issue numbers are rewritten to
CannObserv/observo#N. Where they name Observo's ACL user, group, host or env file (observo,observo.process,observo-primary,/etc/observo/.env), readprocessor,processor.process,co-processor,/etc/processor/.env.
Summary
Observo becomes the cohort's text-extraction processor: consume content.process from the cohort broker, extract with co-core from one source_spec, store the canonical text in the permanent content-addressed store (CannObserv/observo#628), publish content.derived. Issuer and first consumer is Watcher, which stops extracting in-process. Design of record: watcher docs/plans/2026-09-24-observo-extraction-and-diff-design.md — Section 3 is the requirements list this issue mirrors; Section 1 is the contract.
The requirements are the contract; how is Observo's — in particular whether a command mints an asset Job or runs on a lighter path (at Watcher's volume, ≤ ~100 commands/day, lighter is recommended, not required).
Requirements
- Broker membership — ACL user
observo(filed in CannObserv/broker), groupgroup_name(CONTENT_PROCESS, "observo"). - Consumer — at-least-once is tolerated by write-if-absent; undecodable frames to a DLQ; transient errors publish nothing (pending entry reclaimed); deterministic errors publish
processing_failedwithterminal=true. - Input —
gs://co-gcs-blobsvia the co-coreBlobStore(cannobserv#475) under anobjectViewergrant. CannObserv/observo#491'sgs://refusal governs operator-typed job URLs; bus inputs are a separate path and that rule is not extended to them. Verify the bytes hash toinput_digest; mismatch → terminalinput_digest_mismatch. - Extraction — co-core HTML / PDF / CSV extractors from one spec, co-core's media-type dispatch (lifted from watcher in cannobserv#486). The
<0.12pin has to rise, and at the cutover Observo and Watcher must run the same co-core: the extracted bytes must be identical or every fingerprint in Watcher shifts.processor_version = "{co-core version}+{local extraction generation}". - Output — co-core
canonical_textbytes togs://co-gcs-observo/blobs/<sha256>.binthrough CannObserv/observo#628 step 2's write path andblobstable; write-if-absent, never deleted (D1).empty: truewrites nothing. - A scoped read grant for Watcher — CannObserv/observo#628 puts every artifact under
blobs/, so a bucket-wide grant would expose video too. Scope it to derived text (IAM condition, or a dedicated prefix/bucket — your call). The contract states only: read-only, derived text only. - Failure isolation — text extraction runs in its own process, outside the media worker fleet's failure domain (archiver#179's stated counterweight: a GPU/video outage must not stall text derivation).
Non-requirements
No diff computation (Watcher runs difflib over two stored texts at notification time). No source_spec lists. No processor-version pin on the command. No Job API surface for Watcher — the bus is the interface.
Relation to CannObserv/observo#628
Depends on step 2 (the blobs/<sha256>.bin write path). Independent of the recipe ledger (step 5): that is a determinism check, and a changed source is a new recipe there, not a signal — change detection stays Watcher's.
Comment by @gregoryfoster on observo#629, 2026-09-24
cannobserv v0.19.4 carries the contract (tag v0.19.4; CannObserv/cannobserv#486 merged as 4e57ecc4). Per requirement, what it gives you and where it departs from the design's Section 1 table:
Requirement 4 — the pin and the dispatch. co-core[extract]>=0.11,<0.12 → >=0.19.4,<0.20. PdfExtractor.extract(raw, config) is unchanged and tests/shared/test_co_core_contract.py's floor assertion still holds; I found no Breaking CHANGELOG entry in 0.12–0.19.4 naming the extract package, but confirm against this repo's imports rather than take it from me. The dispatch is two halves: from co_core.pure.extract.dispatch import extractor_for_essence (needs the extra; total, HTML for anything unknown) and extraction_overrides_for_essence(essence) (re-exported), merged over extraction_config_from_spec(source_spec). The essence is command.media_type, already resolved by Watcher — there is no media_type_hint / media_type_override pair (deltas D1: the URL-extension tiebreaker needs the origin URL, which you never see). Re-resolving it is idempotent but unnecessary. processor_version(LOCAL_GENERATION) spells the version string.
Requirement 5 — the output. bytes = canonical_text(result.chunks); empty = (bytes == b""), not not result.chunks (deltas D2: PdfExtractor emits a chunk per blank page, so a one-page scanned PDF is one empty chunk, and the issuer's guard is byte-level). output_digest = canonical_text_fingerprint(chunks) (sha256:<hex>); bare_sha256(output_digest) is the store key for uri_for. output_media_type = CANONICAL_TEXT_MEDIA_TYPE (the Emit refuses anything else). spec_fingerprint = spec_fingerprint(source_spec) (send None if it raises), spec_schema_version = spec_schema_version(source_spec). When empty, digest and URI are None and size is 0 — ProcessingCompleteEmit refuses any other combination.
Requirement 3 — the input. input_digest is bare hex; validate with validate_fingerprint and read through BlobStore.get(...), never by resolving input_uri as a path. Bytes that hash differently → input_digest_mismatch; a reference the store does not recognize → invalid_input (terminal; Watcher does not re-fetch); bytes gone → input_unreadable (terminal for the command; Watcher re-fetches, capped); a processor other than "extract" → unsupported_processor (decode the command with the canonical class so you can refuse it). transient is the one non-terminal token, published only if you choose to surface a retry; ProcessingFailedEmit refuses a terminality that disagrees with the token.
Requirement 2 — the bus. Decode with co_core.pure.adapters.bus.envelope.from_wire, emit with ProcessingCompleteEmit / ProcessingFailedEmit; group group_name(CONTENT_PROCESS, "observo") = observo.process; DLQ dlq_name(CONTENT_PROCESS). The producer/consumer live in co-core-aio[bus], which this repo does not declare today (its pyproject says only co-core is imported) — that is a second new dependency alongside the co-core-sync[gcs] CannObserv/observo#628 takes for the store.
Contract: cannobserv docs/CHANGE_BUS.md § "Watcher-issued processing — the content.process pair"; the six departures and why: docs/plans/2026-09-24-content-process-contract-co-core-deltas.md.
Comment by @gregoryfoster on observo#629, 2026-09-24
Broker side is in place for this (CannObserv/broker#62, live on co-broker since 2026-09-24):
- ACL user
observoexists. Root:~content.process ~content.process.dlq,+xreadgroup +xack +xautoclaim +xgroup|create +xlen +xrange +xinfo|stream +info +ping. Selectors:(+xadd ~content.derived ~content.process.dlq),(+xdel ~content.process.dlq). That is what co-core-aio'sAsyncBusConsumerissues (ensure_groupwithMKSTREAM, the blocking read, ack,XAUTOCLAIMreclaim,dead_letter'sXADD) plus the publish, plus the reads and theXDELthe drainer ofcontent.process.dlqneeds - the writer of a queue is its drainer here. Nothing was captured underMONITOR; the first capture is yours to take, and a command the consumer issues that is not granted surfaces asNOPERM(ACL LOGon the broker names it - ask, and it gets widened live, no restart). - Withheld, on purpose:
+xpending/+xclaim(the driver never issues them; granted only if ops tooling runs as this user),+xread,+xtrim,~content.blobs,~content.revisions, and+set/+exists- the broker records that Observo keeps no dedupe key (idempotency is the content-addressed, write-if-absent store). If that changes, the key pattern is inventoried in broker'sdocs/STREAMS.md, Non-stream keys ondb0, before it is written; that table is what keepsnoevictionsafe there. - Group:
group_name(CONTENT_PROCESS, "observo")=observo.process,ensure_groupfrom$. The broker's probe already watches for it: once watcher#325 issues a command and the group is not on the stream, the probe reportsgroup-missingevery ten minutes until it is. - Credential: minted on the broker node; the handoff is
pw OBSERVOthere into/etc/observo/.envas the bus URL's credential (redis://observo:<pw>@broker:6379/0). Human step, tracked on broker#62. - Tailnet:
observo-primaryis not inco-broker's netmap yet, so a policy ruletag:observo-primary->tag:brokeron 6379 is needed in the admin console before the first connection. Also on broker#62.
Two things the broker's inventory marks as yours to state, the way watcher#245 and replicator#19 did for theirs:
NOPERMclassified transient. The broker's whole grant-recovery story is that a too-narrow grant makes a participant back off rather than dead-letter valid work. Thecontent.derivedrow and the ACL header say this is a requirement on the consumer here, not something assumed.- Durability under
maxmemory. The broker refusesXADDinstance-wide at its cap (OOM command not allowed); theProducer durability under OOMcell forcontent.derivedreads unverified. The design's transient rule (publish nothing, let the pending entry be reclaimed) covers it only ifOOMis classified transient - and whether a completed extraction is re-run or only re-published on the reclaim is the same question.
Comment by @gregoryfoster on observo#629, 2026-09-24
The broker credential is in place (CannObserv/broker#62, 2026-09-24): CO_OBSERVO_BROKER_TOKEN in /etc/observo/.env, and a PING as observo from observo-primary succeeded. The broker node now holds only its digest.
Correction to the comment above: the host in the bus URL is 100.97.91.19, not broker. observo-primary runs Tailscale with --accept-dns=false (this repo's docs/reference/tailscale.md), so the MagicDNS name the other services use does not resolve here:
redis://observo:<password>@100.97.91.19:6379/0
The tailnet rule admitting tag:observo-primary to tag:broker is live, and the path is direct at about 1 ms once formed; the first packets after the rule rode DERP. The client-side latency cells in broker's docs/NETWORK-PATHS.md are this issue's to fill, from observo-primary as observo, once the consumer connects.
Activity
gregoryfoster commentedon Sep 29, 2026
Handoff to the
co-processoragentRead the spec first. It is self-contained, and it is already committed here at
docs/specs/2026-09-29-processor-service-design.md.Done (Observo session, 2026-09-29):
co-processor: 8 GB, 20 GB disk, pdx. uv, Python 3.12.3, gh and Claude Code are preinstalled.redis-server(the scratch bus) runs on localhost./etc/processoris at mode 700.tag:processor,--accept-dns=false. The broker at100.97.91.19:6379is reachable (direct path, about 1 ms).~/.ssh/processor_deploy, host aliasgithub-processor. The repo is cloned at~/processor, current withmain.processorfor the content.process consumer; retireobservo's grants broker#75 is filed: ACL userprocessor, stripobservo, probe.Outstanding, human steps:
GH_TOKEN_PROCESSORinto~/processor/.env.pw PROCESSOR) into/etc/processor/.env, after broker#75.co-gcs-processor, service accountco-gcs-processor-writerplus its key, and the grants in spec §2.co-pypi-readerkey for the private package index.Outstanding, drafts for approval (spec §6):
CHANGE_BUS.md.Every external post needs explicit approval, each time.
First task: write
AGENTS.md, run writing-plans for v1, then build with TDD.Hard ordering:
processor.processmust exist before CannObserv/watcher#325 issues its first command.#1 docs: amend spec to co-core 0.19.7, add .gitignore
gregoryfoster commentedon Sep 29, 2026
Review from
co-processor, 2026-09-29I checked the spec and the handoff against co-core
v0.19.4, Watchermain, and the linked issues. The design holds. Five things change or firm up before the build.1. The co-core pin is stale: Watcher is on 0.19.7, not 0.19.4. Watcher's
pyproject.tomlallows>=0.19.6,<0.20, and itsuv.lockresolvesco-core/co-core-aioto0.19.7. At==0.19.4, Processor would report"0.19.4+1"against Watcher's"0.19.7+1"on every shadow command, which is Open Question 1's hazard on day one. The 0.19.5–0.19.7 CHANGELOG entries touch no HTML / PDF / CSV extraction output. They are GCS pointer verbs,signed_url,GcsCopyIfAbsent, stores re-checking the fingerprint against the bytes written (0.19.6), and email / docs fixes. Moving the pin to==0.19.7should therefore be free.co_core.pure.extractandco_core_aio.busare byte-identical fromv0.19.4tov0.19.7, and thecontent.processtypes are unchanged; 0.19.6 only added thecontent.persistpair. The spec is amended to==0.19.7/"0.19.7+1"(D5, §2, §5, §7) in 99ac0e7. 0.20.0 is out, and Processor stays<0.20with Watcher.2. Open Questions 2 and 3 are answered. Q2: no. At 0.19.4,
AsyncBusConsumeroffers only primitives:ensure_group,read,ack,claim_stale/claim_stale_page, anddead_letter. It has no delivery count, no max-deliveries hook, and no run loop, and 0.19.5–0.19.7 add none. The in-memory 3-strike counter stays, and Processor owns its loop. For reclaim it usesclaim_stale_page, which returns malformed frames fordead_letter.claim_staleraises instead. Q3: the overrides win. Watcher's_extract_with_specbuilds{**config_from_spec, **extra_config}, and Watcher's_DEFAULT_EXTRACTOR_MAPis entry-for-entry co-core'sEXTRACTOR_BY_ESSENCE.3. The contract names check out. Every name in the spec's quick-reference table exists at
v0.19.4. One correction to the recreated body: the co-core comment says "read throughBlobStore.get(...)", but the Protocol method isopen(fingerprint), as the spec already says.store(data, fingerprint, media_type).openraisesFileNotFoundErrorfor a missing blob (→input_unreadable).4. The child process (§3 amended). A
ProcessPoolExecutorcannot kill a running task on timeout without private internals. On 3.12 it forks from an asyncio parent that holds a GCS client, and the child inherits that address space, which makesRLIMIT_AShard to size. v1 instead uses a freshspawn-context process per command, killed on timeout, withRLIMIT_ASset in the child. At ≤ ~100 commands/day the start-up cost is noise.5. The parity corpus needs real samples. Watcher's only file fixture for extraction is
tests/fixtures/sample.html, so seeding from Watcher's fixtures proves little. The plan builds synthetic HTML / PDF / CSV cases first. Before shadow, it adds a sample of real watched items (raw digest,source_spec, resolvedmedia_type, Watcher's recorded fingerprint), which needs a small export from Watcher.Standing state. Done: the VM, the scratch Redis, the tailnet path to the broker, the deploy key and clone, the Claude Code login, the GitHub token. Still open:
/etc/processoris still empty.co-pypi-readerkey.Watcher#325 has not started (
mainhas no issuer code), so the hard ordering is not at risk yet.Done in 99ac0e7: a
.gitignorefor.env/.venv/.wheelhouse, and the spec amendments above.Proceeding with:
.wheelhousebuilt from the cannobservv0.19.7tag. This unblocks development before theco-pypi-readerkey arrives. find-links locks by filename, so the mirrored index swaps in later. CI still needs the key.AGENTS.mdand the v1 plan indocs/plans/. Perwriting-plans, the build starts after the plan is reviewed. The pure core and every shell are testable on this VM with the local store, fakes, and the scratch Redis.Suggested: create
processor.processwith oneensure_groupfromco-processoras soon as broker#75 and the credential land, before v1 is deployed. That takes the hard ordering off the build's critical path. Commands queue in the group meanwhile, and Watcher's reaper re-issues the stale ones, which shadow tolerates.The Section 6 drafts come next, one at a time for approval, starting with watcher#325. It carries the pin, the bucket, the reader grant, and Open Question 1, the one open item that could still force a design change.
#1 docs: AGENTS.md and the v1 implementation plan
#1 docs: v1 plan approved; fold in broker#75's handoff order
#1 feat: scaffold, exact co-core pin and its test (plan step 1)
#1 feat: pure extract core with Watcher-generated parity goldens (pla…
#1 feat: killable child runner with RLIMIT_AS and a scrubbed env (pla…
#1 docs: plan and spec record the child as a bare subprocess, not mul…
#1 feat: settings, store builders, transient-error classification (pl…
#1 feat: the handler, spec §4's failure table as code (plan step 5)
#1 feat: the consumer loop on processor.process (plan step 6)
#1 feat: entry points, JSON logs, DLQ CLI, systemd unit, runbook (pla…
#1 fix: CR 1 — stop semantics, DLQ id, settings errors, log contract
#1 fix: CR 2 — child protocol, reclaim batching, settings ranges
11 remaining items
#1 fix: CR 10 — GCP provisioning ran 2026-10-02: rows done, the secti…
#1 fix: CR 11 — the direct-path row carries the 2026-10-02 measuremen…
#1 fix: CR 12 — Watcher's reader grant ran 2026-10-02 (co-gcs-blob-re…
gregoryfoster commentedon Oct 2, 2026
Shipped in #14 (FF to
mainat7a5b87a,60aedc8..7a5b87a): the GCP provisioning runbook, run by the operator 2026-10-02; the writer's preflight reached both buckets. Watcher's reader grant (co-gcs-blob-reader) is live and reported on CannObserv/watcher#325. The direct-path prerequisite moved to #15. CI green on run 37043235931. CR 10–12 are in the range.#1 fix: CR 17 — a repeatable smoke test: scripts/smoke_scratch_bus.py…
#1 fix: CR 18 — spec §6 step 3 done: the scratch-bus smoke test passe…
#1 fix: CR 19 — the 65 MB is the unit's cgroup MemoryCurrent, not res…
#1 fix: CR 17 — the refusal test names a TEST-NET address, not the br…
gregoryfoster commentedon Oct 3, 2026
Live. Shipped in #22 (FF to
main,6e518d8..02ac0f4).processorunit is installed and enabled onco-processorfrom6e518d8:starting, thenconsuming, 0 restarts.scripts/smoke_scratch_bus.py(CR 17,f295eb2) runs one real-corpus command through the real Consumer on the scratch Redis (db 14, never the broker) against the production output bucket.f3f1071).b8f6d0f1….bin, Watcher's recorded text for that revision. Repeat runs are write-if-absent no-ops.Next: watcher#325 ships and sends its first shadow command (cutover step 4). That gives the first real
content.derivedfact and broker#75's first ACL capture. Open: #2 (containment), #5 (SocratiCode), #15 (reboot re-check), #20 (spec wording), #21 (processor_error, blocked on cannobserv).gregoryfoster commentedon Oct 3, 2026
Processor's answers to Watcher before the shadow flip (2026-10-03). These were asked live by watcher-agent (CannObserv/watcher#325), and this comment is Processor's durable record of them.
co-processor), strictly serial, stream order with one exception. An attempt can end with no fact and no ack: a transient error (uncapped), strike 1–2 of 3, or a restart mid-command. That entry stays in the PEL while later commands are answered. It is re-run only once it has been idle forreclaim_min_idle_ms(600 s), on the next 60 s walk. Watcher's re-issue rule ("re-issue once a later command is answered") can therefore re-issue a command that Processor will still run about 10–11 min later. That costs duplicate work, never a wrong result: Watcher discards the superseded fact, and the store is write-if-absent. Settled: Watcher's stuck threshold is 1800 s (WATCHER_PROCESS_COMMAND_TIMEOUT_SECONDS, the default), so a single failed attempt never triggers a re-issue. The only residual is a command that fails transiently for more than 30 min. In that case the original's fact lands on an expired row and is dropped aslate.input_digest, butinput_urimust equalgs://co-gcs-blobs/blobs/<hex>.binexactly, or the command gets a terminalinvalid_input. The real corpus covers Watcher's 4 production items across 9 revisions, alltext/html. Open with Watcher: 6 of the 9 exported specs have notarget.targetdoesn't affect the extracted text, butspec_fingerprint(co-core 0.19.7) depends on it. Settled: the comparator checksoutput_digest, plusspec_fingerprintwhen both sides have one. Both sides fingerprint the same spec dict (3 of Watcher's 4 live items have notarget), so this can't cause a mismatch.target. Processor reports thatcommand_id's disposition,output_digestand timings from the journal. Watcher owns the plan on watcher#325. If co-processor reboots, a new session runs Tailnet path to the broker: DERP-relayed at idle, direct only under traffic (NAT hairpin) — confirm it holds for the running service #15's post-boot checks before the flip. The reboot (Tailnet path to the broker: DERP-relayed at idle, direct only under traffic (NAT hairpin) — confirm it holds for the running service #15, epic Burn-down: open issues after v1 go-live (done 2026-10-11; next: #37; parked: #54) #24) is preferred before the flip, but it isn't required.gregoryfoster commentedon Oct 3, 2026
Shadow is live, and the first real command matched (2026-10-03). Spec §6 cutover step 4 has begun.
WATCHER_EXTRACT_MODE=shadowand restarted. That came after Tailnet path to the broker: DERP-relayed at idle, direct only under traffic (NAT hairpin) — confirm it holds for the running service #15's reboot of co-processor (16:53Z, clean), as the go-live plan on watcher#325 ordered.01M41H10MS8H92WXP2V8JTH0F0, from the operator's Check now on watched item01KRZ2DV43YCKB90AB5PFM55RD(spec_index 0,text/html).ack: complete, attempt 1, stream entry1791052776110-0, logged at 18:39:37.077Z. Timings: read 292.5 ms, extract 544.7 ms, store 122.3 ms, publish 2.0 ms, total 962.2 ms. This is the first command through the production broker, so the "no real-broker command yet" debt is paid.output_digestsha256:b8f6d0f1…00bdfequals Watcher's local fingerprint, andspec_fingerprintequals too. Size is 17327 bytes,processor_versionis0.19.7+1. Verdict: match, with no mismatch audit.Left for step 4/5:
targeton watcher#325, as their scheduled fetches run.gregoryfoster commentedon Oct 4, 2026
NOPERM on the fact publish: shown in production, 2026-10-04 (a rehearsal with broker-agent; broker's record is CannObserv/broker#75 issuecomment-5976099134, CannObserv/broker f3783a1).
The broker narrowed
processorso thatXADD content.derivedwas refused and the DLQ, XREADGROUP and XACK stayed allowed. The operator pressed Check now on one item, and then the broker restored the grant.01M42DS2W8FX1HEXEPG6BPSGA3delivered (entry1791082924951-0)leave_pending: transient, attempt 1,publish failed: NoPermissionError: …ack: complete(reclaim re-run 658 s after delivery; publish 1.3 ms)The broker's side:
processing_completefor the command, which Watcher consumed.Our side:
NRestarts=0.This confirms spec §4 as built:
OOM on the publish stays shown on scratch only, by agreement: a production rehearsal would refuse every participant's writes.
gregoryfoster commentedon Oct 6, 2026
Cutover happened: Watcher switched to
WATCHER_EXTRACT_MODE=processorat 2026-10-06T21:35:24Z (CannObserv/watcher#326). Recorded from epic #24's check-in.01KRZ2DV43YCKB90AB5PFM55RD, at 21:07:03Z. On Processor's side, inputfea1c587…gave outputsha256:2d98c2c4…(17,360 bytes), the first new output since go-live, and Watcher's local extraction agreed.ack: completehere,check.no_changewithsource: processoron Watcher's side.428af0e2a016, containmentrequired). It was healthy before and after.Remaining before this closes: Watcher's soak, then its deletion of the local path, the mode variable and
co-core[extract](watcher#326). Spec §6's cutover order then marks step 5 done.shadowthenlocal, plus a restart.gregoryfoster commentedon Oct 10, 2026
Closing: Processor v1 is the cohort's text extractor, and the only one.
processor, groupprocessor.process: broker#75 (closed 2026-10-05). The group was created 2026-10-01T22:09:41Z, ahead of Watcher's first commandcontent.process.dlq; transient errors publish nothing and stay pending; deterministic errors publish terminalprocessing_failed. The cap and dead-letter path are #17 and #28gs://co-gcs-blobsread-only, andinput_digestis verified0.19.7+1on both sides, 87/87 shadow matches). Since #47 (#49), Processor owns the pin (D5 restated)gs://co-gcs-processor/blobs/<sha256>.bin, write-if-absent, never deleted (D4)objectVieweris the scoped grant (D4). Watcher has read it in production since the switchco-processor. Extraction runs in a killable child under Landlock and seccomp (#2)Cutover (spec §6, now marked done in #49):
processorat 2026-10-06T21:35:24Z;Operations: live since 2026-10-02 from read-only releases (#2, #34), with the drift check (#35) and liveness (#39) reporting to Status.
Follow-ups: the remaining backlog is on epic #24 (#5, #29, #37). #47's Watcher notice is tracked on #47.