ID-89 RESEARCH — flow.py refactor feasibility: cocoindex-native vs hand-rolled

ID-89 RESEARCH — flow.py refactor feasibility: cocoindex-native vs hand-rolled

Artefact: {89.1} RESEARCH — the only spec-chain artefact authorised for Task ID-89 (RESEARCH-only Task; no PRODUCT/TECH/implementation without explicit ratification). Status: Awaiting Liam ratification at the spec gate. No code, migrations, or ledger writes were made authoring it. Date: 07/06/2026. Author: Task Planner (Opus 4.8, 1M context) — fresh dispatch, read-only. Trigger: Liam, S321 — scripts/cocoindex_pipeline/flow.py is 3,469 lines (tokenize count, which the executable-line arithmetic uses; wc -l reports 3,468 — tokenize includes the final line); cocoindex’s flow model is core to what the library provides; suspicion that KH hand-rolled machinery cocoindex 1.0.x gives us for free. Pinned version verified: cocoindex[postgres]==1.0.3 (requirements.txt:49); cocoindex.__version__ == "1.0.3" confirmed at runtime 07/06/2026. Latest upstream: 1.0.7 (PyPI, checked 07/06/2026).

---

0. Headline verdict (read this first)

1. The suspicion is largely NOT confirmed. flow.py does NOT re-implement cocoindex’s flow model — it sits ON it: @coco.fn(memo=True) components, coco.mount_each fan-out, localfs.walk_dir + a custom protocol-conforming URL source, mount_table_target(managed_by=USER) + declare_row, ContextKey + @coco.lifespan, ops.entity_resolution.resolve_entities, LiteLLMEmbedder, RecursiveSplitter. Every one of those is the native construct, used the native way.
2. What makes flow.py big is what cocoindex CANNOT do — empirically proven across id-28 §R-series, the S297 write-model bug ledger (A–G), and the {75.16}/{75.17} live-engine probes: no per-row/per-stage completion callbacks, no run op_id, no retry primitive, no cross-target transaction/ordering, no ContextVar propagation across the _LoopRunner daemon thread, no LLM-extraction op (ExtractByLlm ABSENT). The KH observability/rollup/error-classification substrate (~730 lines) and the engine-contract workarounds exist because the engine has no surface for them. None of these gaps close in 1.0.4–1.0.7 (changelog-verified §3.3).
3. Half the file is not code. Of 3,469 lines, ~1,411 are executable; 947 comment lines + 737 docstring lines (49%) are an evidence ledger of live-proven engine behaviour (tokenize/ast count, 07/06/2026). The “3,500-line monolith” framing overstates the code mass by ~2.5x.
4. There ARE genuine native-adoption candidates — all small, all probe-gated (§5 Options 2a–2e): TableSchema.from_class + dataclass rows, scoped coco.exception_handler for per-item containment, engine reconciliation possibly subsuming _trim_stale_form_fields, coco.map for the chunk loop. None reduces flow.py by more than a few hundred lines.
5. The single biggest refactor risk is the memo fingerprint. The engine fingerprints memoised components on module + qualname + logic tracking ({75.8} verdict: memo_fingerprint._canonicalize_dataclass = ("dataclass", module, qualname, field-values); logic_tracking='full' is the @coco.fn default — empirically verified on the installed pin). Moving or renaming any @coco.fn(memo=True) component invalidates its memo → next walk re-runs Stage-3 Anthropic extraction for the whole corpus. Any mechanical decomposition therefore needs a planned full-reprocess window + burn budget. This is the {56.18}-flagged memo-fingerprint OQ, now load-bearing.
6. Recommendation: Option 1 (mechanical decomposition of the observability + schema substrate into satellite modules, behind a planned reprocess window) is the only option whose value clearly exceeds its risk, and even that is comfort-grade, not functional. Options 2a–2e are cheap probes worth queueing opportunistically. Option 4 (full native rewrite) is a NO-GO. See §7 options table.

---

1. Code-intelligence orientation (per .gitnexus/CLAUDE.md “Always Do” / Inv 2)

gitnexus_query({query: "cocoindex flow ingestion pipeline", repo: "knowledge-hub"}) returned "processes": [] (zero indexed execution flows — expected: gitnexus processes are TS-centric and the Python pipeline is engine-invoked, an edge gitnexus does not trace through the framework). Definitions surfaced verbatim:

• "id": "Function:scripts/cocoindex_pipeline/flow.py:app_main", "startLine": 2864, "endLine": 3387, "module": "Cocoindex_pipeline"
• "id": "Function:scripts/cocoindex_pipeline/flow.py:ingest_url", "startLine": 2338, "endLine": 2414, "module": "Cocoindex_pipeline"
• "id": "Function:scripts/cocoindex_pipeline/flow.py:_build_db_ctx", "startLine": 906, "endLine": 928

gitnexus_query({query: "url source feed ingest"}) surfaced the TS-side seam (proc_136_get "GET → CreateClient", proc_142_post "POST → CreateClient" over app/api/intelligence/workspaces/[id]/... routes and lib/intelligence/health.ts:getPipelineHealth at proc_138_get step 2) plus flow.py definitions _log_ssrf_rejection (2263–2302), _backlink_feed_articles (2312–2334), _ingest_url_body (2417–2592).

gitnexus_query({query: "stage 5 entity resolution"}) surfaced Function:scripts/cocoindex_pipeline/stage_5.py:_run_stage_5_resolution (startLine: 203, endLine: 544) and flow.py’s _classify_stage_exception (264–381), _emit_workspace_unmapped_warn (486–521), _empty_stage_counts (704–721).

gitnexus_context({name: "app_main", file_path: "scripts/cocoindex_pipeline/flow.py"}): "incoming": {} (zero indexed callers — the component is registered via coco.App(AppConfig(name="kh_pipeline"), app_main) and invoked by the engine, so no risk-verdict field was emitted; caller count = 0, "processes": []). Outgoing calls (19): _run_stage_5_resolution, _classify_stage_exception, _pydantic_error_detail, _redact_error_message, _emit_stage_error_log, _empty_stage_counts, _emit_pipeline_run_webhook, load_workspace_manifest, FeedUrlSource.items, _FlowRetryCounter.get, _FlowStageCounter.get (×7), _FlowTaxonomyMissCounter.tally_by_field, _FlowItemFailureCounter.tally.

gitnexus_context({name: "ingest_url", file_path: "scripts/cocoindex_pipeline/flow.py"}): incoming calls = bound_ingest_url only (the named closure in app_main); outgoing = _ingest_url_body; "processes": []. Blast radius for both seam symbols is LOW on the indexed graph — the real coupling is engine-mediated (mount_each / memo / reconciliation), which is exactly why §6’s engine contracts, not the call graph, bound any refactor.

ccc was not invoked: gitnexus surfaced the relevant symbols in all three queries (the brief’s ccc fallback condition — “where gitnexus doesn’t surface relevant symbols” — was not met). flow.py is Python, so ast-dataflow does not apply; symbol inventories below are from grep sweeps (grep -n "^def \|^class \|^@" scripts/cocoindex_pipeline/flow.py) and AST parsing of all satellite modules, per the dispatch brief.

---

2. Inventory — what flow.py and its satellites actually do

2.1 flow.py (3,469 lines) by region

Region (lines)	~Lines	Responsibility	Cocoindex surface used
1–165	165	Module docstring (1.0.3 API-deviation ledger) + imports (via _coco_api façade)	—
170–199	30	_emit_upsert_log — per-row UPSERT log substrate, unwired (no per-row callback exists, id-28 §R8)	none available
202–523	320	Inv-25/26 error machinery: _PIPELINE_ERROR_CLASSES (6-class vocabulary mirrored to TS), _EntityResolutionStageError, _classify_stage_exception, _pydantic_error_detail, _redact_error_message (PII: input_value + UUID redaction, bl-165), _emit_stage_error_log, _emit_workspace_unmapped_warn	none (engine has exception_handler, but no classification/PII/webhook)
525–901	375	Per-flow counters (_FlowRetryCounter, _FlowStageCounter, _FlowTaxonomyMissCounter, _FlowItemFailureCounter), _item_failure_branch, _empty_stage_counts (7-stage Zod-enforced map), extraction recorders, _emit_pipeline_run_webhook (Inv-16/17/18 rollup → Vercel pipeline_runs route)	none (no per-stage callbacks §R8, no run op_id §R9)
903–932	30	_build_db_ctx / DB_CTX — ContextKey with duplicate-registration defence (ID-49.7 dual-import hazard, largely historical post-{67.2})	coco.ContextKey
935–1123	190	Stage-4 embedding: LiteLLMEmbedder lazy singleton, bl-223 token-budget truncation (_truncate_embedding_input, tiktoken), embed_content_text dim assertion, _encode_pgvector text-literal encoder	cocoindex.ops.litellm.LiteLLMEmbedder
1125–1436	310	Six hand-written TableSchema declarations (content_items, q_a_extractions, source_documents, reference_items, entity_mentions, content_chunks) + two form schemas + MIME maps + FORM_WRITE_GRACEFUL_EMPTY_REASON	TableSchema / ColumnDef (postgres connector)
1439–1489	50	_build_dsn — explicit COCOINDEX_DB_DSN, pooler-host regression guard (ID-49.8 NXDOMAIN crash)	—
1494–1574	80	_field / _stamp_if_model shape-agnostic helpers, _KH_PIPELINE_DOC_NS uuid5 namespace, _WORKSPACE_MANIFEST_FILENAME, _to_source_relative ({66.22}/BUG-A absolute-path normalisation)	—
1576–1888	310	ingest_file (memoised per-item component) + _ingest_file_body dispatcher: {66.19} explicit-kwarg threading + daemon-thread ContextVar re-bind; {80.8} forms/content fork via resolve_route; bl-219 unmapped soft-warn	@coco.fn(memo=True)
1891–2182	290	_ingest_content_branch — Stage 2 (adapter call) → Stage 3 (3 extractor awaits) → Stage 6 declare_rows (sd/ci/qa/em) + Stage 4 embed + ID-56.8 chunking ({66.16}/BUG-F natural-key dedup for entity_mentions)	declare_row, RecursiveSplitter
2184–2334	150	URL helpers: _url_filename, _url_is_pdf (D-12 HEAD sniff), _fetch_url_bytes, _pullmd_extraction_method, _log_ssrf_rejection (D-9 quality-log row), _backlink_feed_articles (D-7 raw-pool UPDATE)	coco.use_context
2338–2593	255	ingest_url + _ingest_url_body — WP-C steps 1–8: SSRF gate, PDF/HTML route, sd+ri declare pair, {75.17} FK-tolerant backlink (constraint-name string compare)	@coco.fn(memo=True), declare_row
2596–2860	265	_ingest_form_branch (Path-B form write: suffix guard, analysis_failed row, Inv-17 graceful-empty, {80.13} coerce_extracted_form memo-HIT normalisation) + _trim_stale_form_fields (Inv-16 shrink DELETE)	declare_row
2865–3389	525	app_main: idle-mode gate, manifest load-or-abort, in_progress webhook, 8× mount_table_target, walk_dir + mount_each(ingest_file) with bound_ingest_file containment closure, FeedUrlSource + mount_each(ingest_url), Stage-5 post-pass call, flow-scope except/finally rollup + terminal webhook	mount_table_target, walk_dir, mount_each, component_subpath, use_context
3391–3468	78	_register_pg_codecs (jsonb pool codec, {66.16}/BUG-D), kh_pipeline_lifespan (asyncpg pool env-scope), KH_PIPELINE_APP = coco.App(...)	@coco.lifespan, EnvironmentBuilder.provide, coco.App

Comment-density evidence (tokenize + ast, 07/06/2026): total 3,469 lines; 947 comment lines; 737 docstring lines; 374 blank; ~1,411 executable. The prose encodes the S297 bug ledger, the {66.x}/{75.x}/{80.x} live findings, and the 1.0.3 API-deviation catalogue — it is institutional memory, not noise.

2.2 Satellite modules

Module	Lines	Responsibility	Native vs KH
_coco_api.py	137	PEP 562 lazy façade over the 12 off-surface cocoindex symbols; one-file fix on a version bump; preserves bare-MagicMock test collection-safety	KH insulation over native symbols
adapters.py	336	Stage-2 per-MIME adapters: convert_binary_to_markdown (docling/passthrough; HTML branch retired {75.9}), _pullmd_fetch (epoch-keyed memo, D-4), extract_source_provenance	@coco.fn(memo=True) wrapping KH HTTP/docling calls
flow_context.py	300	Five ContextVar bind/current pairs (flow meta, retry, stage, manifest, taxonomy-miss counters)	KH — exists BECAUSE the engine does not propagate ContextVars across _LoopRunner ({66.19}); retained for in-task unit-test callers
extraction.py	1,061	Pydantic extraction shapes (+Stamped variants), Path-A extractors (extract_classification / extract_qa_form / extract_entity_mentions, all @coco.fn(memo=True)), _anthropic_retry (tenacity, Inv-23), prompt caching, classify_pydantic_error, truncation guard	KH — and this IS the native pattern: 1.0.3 ships no LLM-extraction op (ExtractByLlm/LlmSpec/LlmApiType ABSENT, OQ-3/S256); upstream’s own documented pattern is “call your LLM client inside @coco.fn(memo=True)”
stage_5.py	545	Stage-5 post-pass: op-scoped SQL selects (run mentions, canonical roster, prior op-key holders, aliases) + resolve_entities invocation + canonical_name UPDATEs	KH SQL plumbing around the native ops.entity_resolution.resolve_entities
entity_embedder.py / pair_resolver.py	106 / 236	KhEntityEmbedder, KhPairResolver	KH implementations of the documented resolve_entities(embedder=, resolve_pair=) plug-in points — native extension surface
url_source.py	226	UrlItem frozen dataclass + FeedUrlSource/_PassedUrlItems (LiveMapView-conforming snapshot over feed_articles WHERE passed; {75.16} update_all-then-mark_ready watch)	KH custom source on the native runtime-checkable protocol ({75.1} §4 verdict: real, public, supported extension point)
url_normalise.py / url_validation.py	96 / 119	D-8 normalisation; BI-21 SSRF gate (TS port)	Pure KH domain logic
workspace_resolver.py	294	Manifest schema + resolve_route / resolve_workspace (Path-B fork input)	Pure KH domain logic
canonicalisation.py / entity_context.py	49 / 42	Per-doc entity canonicalisation; context snippets	Pure KH domain logic
server.py	705	aiohttp wrapper: /healthz, /stage, bearer-gated /walk (one-shot update_blocking(live=False)), lifespan-only boot via coco.start_blocking() (ID-83 burn guard), worker-thread management	KH operational surface deliberately replacing the native CLI/live mode (bl-221: boot never walks)
prompts.py	180	LLM instruction constants	Pure KH

Total pipeline package: 8,326 lines, of which flow.py is 42%.

---

3. cocoindex 1.0.x native feature map

Sources: the local cocoindex skill (.claude/skills/cocoindex/SKILL.md + references/api_reference.md, invoked via the Skill tool per the dispatch brief), https://cocoindex.io/docs, github.com/cocoindex-io/cocoindex releases, and direct import-and-call probes against the installed cocoindex==1.0.3 (§8 verification block — every symbol below is empirically PRESENT on the pin unless marked otherwise).

3.1 What 1.0.3 provides natively (verified PRESENT)

Capability	Surface	KH usage today
App/flow composition	coco.App(AppConfig(name=...), main_fn), update()/update_blocking()/drop()	USED (KH_PIPELINE_APP; update_blocking(live=False) per /walk)
Processing functions + memoisation	@coco.fn(memo=, version=, logic_tracking=, batching=, max_batch_size=, runner=, memo_key=, deps=)	USED (10 KH-authored memoised components across flow/adapters/extraction, plus the engine-side memoised LiteLLMEmbedder.embed) — batching, version, memo_key, deps UNUSED
Component mounting	mount, mount_each, use_mount, mount_target, component_subpath, coco.map	mount_each + component_subpath USED; coco.map UNUSED (chunk loop is sequential)
Sources	localfs.walk_dir(live=), S3/GDrive/Kafka/postgres PgTableSource/RowFetcher.items(key=)	walk_dir USED; PgTableSource NOT used (hand-rolled FeedUrlSource instead — see Option 2e)
Custom sources	runtime-checkable LiveMapView / LiveMapFeed protocols	USED (FeedUrlSource) — the {75.1} §4 verdict: a real, supported extension point
Targets / declarative state	postgres.mount_table_target(...) → TableTarget.declare_row / declare_vector_index / declare_sql_command_attachment; ManagedBy.USER row-only mode; reconciliation GC of non-redeclared rows	declare_row USED ×8 tables; declare_vector_index deliberately AVOIDED (DDL-via-CLI rule); declare_sql_command_attachment unexplored; reconciliation GC observed live ({80.13})
Schema declaration	hand-built TableSchema(columns={...}) or await TableSchema.from_class(dataclass, primary_key=[...]); ColumnDef(encoder=) hook; Annotated[NDArray, EMBEDDER] vector-dim inference	Hand-built dicts USED; from_class NOT used (Option 2a)
Shared resources	ContextKey[T] (+detect_change), @coco.lifespan, EnvironmentBuilder.provide/provide_with/provide_async_with, use_context	USED (DB pool)
Exception handling	builder.set_exception_handler(...) (global) + async with coco.exception_handler(...) (scoped); ExceptionContext (env_name, stable_path, processor_name, mount_kind, …)	NOT used — per-item containment is hand-rolled try/except closures (Option 2b)
ID generation	resources.id.generate_id / generate_uuid / IdGenerator / UuidGenerator	NOT used — KH uuid5(_KH_PIPELINE_DOC_NS, seed) is deliberate (§6 C-8)
Text ops	RecursiveSplitter (byte budgets, language=), SeparatorSplitter, detect_code_language	RecursiveSplitter USED (bare; language= evaluated and NO-GO’d at {56.18})
Embedding	ops.litellm.LiteLLMEmbedder (memoised .embed), ops.sentence_transformers	LiteLLMEmbedder USED
Entity resolution	ops.entity_resolution.resolve_entities(entities, *, embedder, resolve_pair, is_existing_canonical, existing_policy, on_resolution, max_distance, top_n)	USED (stage_5.py with KH embedder/resolver plug-ins)
Live mode	app.update(live=True), walk_dir(live=True) watch	AVAILABLE but architecturally REJECTED (bl-221/ID-83 burn guard: boot never walks; one-shot /walk only)
CLI	cocoindex update/drop/ls/show	REPLACED by server.py (operational choice, not a gap)

3.2 What 1.0.3 does NOT provide (the gaps that shaped flow.py — all empirically proven)

Gap	Evidence	KH compensation
Per-row UPSERT completion callback	flow.py:31 module docstring; id-28 §R8	_emit_upsert_log substrate (unwired) + _bump("postgres_upsert") at declare sites
Per-stage completion callbacks / run rollup / run op_id	id-28 §R8/§R9	The whole counter + webhook substrate (~730 lines): KH mints run_op_id = uuid.uuid4() per walk, stamps it as a plain row field
Retry primitives (@coco.fn has no retries/backoff kwargs; none on postgres connector)	flow.py:30/34; unchanged through 1.0.7	_anthropic_retry (tenacity) in extraction.py + _FlowRetryCounter
LLM extraction op (ExtractByLlm / LlmSpec / LlmApiType)	ABSENT — OQ-3/S256, knowledge-hub-archive/audits/cocoindex-1.0.3-extractbyllm-spec-reality-investigation.md (cold-storage archive)	Path A: KH extractors calling the anthropic SDK inside @coco.fn(memo=True) — which matches upstream’s own documented LLM pattern (instructor/litellm inside a memoised fn)
Cross-target transactions / write ordering / FK safety	write-model §1: each target on its own autocommit pooled connection, Rust-core ordering, parent-not-before-child observed	FK drops (S297 migration) + integrity-by-construction uuid5 PKs + {75.17} FK-tolerant backlink with walk-2 convergence
.transform() flow-scope chaining	”fictional in 1.0.3” — id-28 §R2/§R3	Plain awaits inside component bodies
ContextVar propagation to per-item components	{66.19}: _LoopRunner daemon thread does not copy the binder task’s snapshot	Named-closure explicit kwargs + local re-bind; flow_context.py retained for test callers
functools.partial as mount_each fn	{66.16}: engine requires __name__ + __qualname__	Named nested coroutines (bound_ingest_file / bound_ingest_url)
Typed memo-HIT returns	{80.13}: memo HIT returns a plain dict (return hint resolved to Any)	coerce_extracted_form normalisation
jsonb / pgvector encoding	write-model R2; ColumnDef.encoder unset on the USER upsert path	Pool-level jsonb codec (_register_pg_codecs); _encode_pgvector ColumnDef encoder
ON CONFLICT beyond the PK	write-model R3	Python-side natural-key dedup + natural-key PKs (BUG-F)
Accurate FileLike docs	write-model R4: size()/content_fingerprint() async; absolute paths despite docstring	await fixes + _to_source_relative

3.3 The 1.0.4–1.0.7 delta (pin-bump assessment, changelog-verified 07/06/2026)

1.0.4: Julia splitter support; column-name validation / SQL-injection hardening in postgres+sqlite connectors; memo cache shared across script and CLI invocations; state methods through dataclass/pydantic fields. 1.0.5: pgvector ext install to default schema; Rust workspace unification. 1.0.6: Elm splitter; better Rust→Python error context. 1.0.7: coco.auto_refresh + unified error-handling channel for live components; coco.stats_group() scoped statistical reports; Apache Iggy connector; LanceDB in-place column adds; postgres halfvec opclasses + NUL-stripping in text/JSONB writes; COCOINDEX_LOG_LEVEL.

Notable for ID-89: no per-row callbacks, no retry primitives, no transaction/ordering options across any of 1.0.4–1.0.7. The three biggest hand-rolled regions would survive a bump intact. stats_group() MIGHT partially overlap the stage-counter substrate and the 1.0.7 “unified error-handling channel” MIGHT overlap the containment closures — both unverifiable from changelog text alone (probe-gated, §5 Option 3). The 1.0.4 shared memo cache interacts with the burn model’s LMDB-clear runbook step and MUST be re-probed before any bump.

---

4. Free-vs-hand-rolled classification

Classification key — N: native, used natively. W: hand-rolled WORKAROUND for a proven engine gap (cocoindex cannot do it). D: KH domain logic (no engine claim). R: re-implemented where a native alternative EXISTS (refactor candidate). A: native surface deliberately AVOIDED for an architectural reason (not a gap).

Responsibility	Class	Notes
Flow composition (App, app_main, mounting, memo, incremental skip)	N	Native, used per the engine’s own idioms
Source walk (localfs) + custom URL source	N	FeedUrlSource implements the supported protocol; {75.16} fixed it TO the proven contract
Row UPSERT targets + idempotent re-ingest	N	declare_row + deterministic PKs; engine reconciliation GC live-observed
Stage-3 LLM extraction (extraction.py)	W	No native op in 1.0.3; KH’s shape IS the upstream-documented pattern
Run rollup / webhook / op_id / counters (~730 lines)	W	§R8/§R9 gaps; persists through 1.0.7
Error classification + PII redaction + stage-error logs	W(+R edge)	Vocabulary/PII/webhook are KH-only; the raw catch-and-contain mechanics MIGHT ride coco.exception_handler (Option 2b)
Per-item containment closures (bound_ingest_*)	R?	coco.exception_handler PRESENT in 1.0.3; containment-vs-observe semantics unprobed
Retry (_anthropic_retry, _FlowRetryCounter)	W	No engine retry primitive through 1.0.7
ContextVar threading (flow_context.py + closures + re-binds)	W	Daemon-thread boundary; no native alternative
Cross-target FK/ordering workarounds (uuid5 integrity, FK-tolerant backlink)	W	Write-model R1; no native transaction surface through 1.0.7
jsonb pool codec, pgvector encoder	W	Engine passes raw values; codec/encoder are the sanctioned hooks
Hand-written TableSchema dicts	R	TableSchema.from_class(dataclass) PRESENT in 1.0.3 (Option 2a)
_trim_stale_form_fields shrink DELETE	R?	Engine reconciliation GC’d non-redeclared rows at S316 — the native model may already cover the shrink case (Option 2c, probe-gated)
Sequential chunk-embed loop	R	coco.map (concurrent fan-out) PRESENT and unused (Option 2d)
Hand-rolled feed_articles enumeration in FeedUrlSource	R	PgTableSource/RowFetcher.items(key=) PRESENT ({75.1} §4.1 called it “almost exactly the enumeration Option A needs”) — but the live-proven {75.16} watch contract argues KEEP (Option 2e)
uuid5 PK scheme vs resources.id	A	KH uuid5 is integrity-by-construction post-FK-drop AND the TS app joins on it; native IDs are component-path-scoped — switching re-keys the corpus (§6 C-8)
declare_vector_index	A	Out-of-band CREATE INDEX not gated by ManagedBy.USER → violates DDL-via-CLI (ID-49.2 journal)
Live mode / auto_refresh / CLI	A	bl-221/ID-83 burn guard: boot never walks; bearer-gated one-shot /walk only
SSRF, URL normalise, workspace resolver, canonicalisation, prompts, form readers	D	Domain logic; no engine claim to replace it
server.py HTTP wrapper	D/A	Operational surface; replaces CLI by design

Net classification: ~45% of flow.py’s executable code is W (gap compensation), ~30% N/D (native usage + domain logic at the seams), ~15% A (deliberate avoidance), and only ~10% R (genuine native-adoption candidates) — and every R item is small.

---

5. Refactor options (risk / effort / value)

Option 0 — Do nothing

Leave flow.py as is. Effort: none. Risk: none functional; ongoing cost is navigation/comprehension overhead and merge-conflict surface on a 3.5k-line file that nearly every pipeline Task touches. Value: zero, but the status quo is live-proven on both envs (ID-88 closed with HTTPS ingress + 202 nudge proof).

Option 1 — Mechanical decomposition, no engine-semantics change (RECOMMENDED shape)

Split flow.py along the §2.1 region boundaries into satellites that already have precedent (stage_5.py, adapters.py): observability.py (counters + webhook + error classification + redaction, ~700 lines), schemas.py (TableSchemas + encoders + MIME maps, ~350 lines), ingest_url.py (URL component + helpers, ~400 lines), leaving flow.py ≈ ingest_file dispatcher + app_main + lifespan (~1,400 lines incl. prose). Effort: S–M (1–2 sessions incl. test moves). Risk: MEDIUM, dominated by ONE hazard: any moved/renamed @coco.fn(memo=True) component or changed component_subpath invalidates memo fingerprints and component identity → full-corpus Stage-3 re-extraction on the next walk + reconciliation churn on target states. Mitigations: (a) move only NON-memoised helpers (counters, webhook, schemas, redaction are all plain functions/classes — the memoised components stay put), (b) keep ingest_file/ingest_url/app_main and all @coco.fn bodies in flow.py, (c) if a memoised component must move, schedule it with an LMDB-clear + full-reprocess window (the burn runbook already exists, write-model §6). Variant (a)+(b) reduces the memo-fingerprint risk to ~zero IF logic_tracking='full' does not fingerprint across module boundaries into moved callees — this is the load-bearing unknown; probe before PLAN (cheap: move a helper in a branch, hash-compare memo keys on the real engine, {75.8} probe pattern). Value: comprehension + merge-surface relief only.

Option 2 — Targeted native-adoption slices (each independently probe-gated)

• 2a. TableSchema.from_class + dataclass rows. Replace 8 hand-dict schemas with typed row dataclasses. Effort S–M. Risk MEDIUM: must preserve ColumnDef encoders (pgvector), OMITTED columns (GENERATED ALWAYS content_text_hash; PG-defaulted created_at family), nullable parity, and the exact column set (any drift = live write failures of the S296 class). Value: type-safety, schema-drift surfaced at import. Verdict: genuine but cosmetic; the OMIT semantics make from_class awkward — probe whether omitted-field control exists before speccing.
• 2b. coco.exception_handler for per-item containment. Replace try/except in bound_ingest_* with scoped handlers. Effort S. Risk MEDIUM-HIGH: UNKNOWN whether a scoped handler CONTAINS (swallows, batch continues) or merely OBSERVES; the {75.16} lesson (BI-19: a faulted component’s declared rows are DISCARDED) means containment semantics are load-bearing. Branch attribution (forms/content/url tally) still needs KH code. Value: marginal — removes ~40 lines. Verdict: probe-then-likely-skip.
• 2c. Engine reconciliation vs _trim_stale_form_fields. S316 live evidence shows the engine GCs rows a component previously declared but no longer declares — which is exactly the Inv-16 shrink case the trim DELETE hand-rolls. Effort XS. Risk LOW-MEDIUM (probe: re-ingest a form with fewer fields, trim disabled, assert engine removes the trailing ftf rows; also covers rows pre-dating engine state — the trim handles those, reconciliation may not). Value: −30 lines + one fewer raw-SQL seam. Verdict: best effort-to-value ratio of the 2x family.
• 2d. coco.map for the chunk-embed loop. Concurrent chunk embedding. Effort XS. Risk LOW (ordering is positional; embeds are independent; watch rate limits). Value: latency only.
• 2e. PgTableSource replacing FeedUrlSource internals. Effort M. Risk HIGH for near-zero value: url_source.py is 226 lines, live-proven through the {75.16} probe matrix, and carries KH grouping semantics (normalise_url collapse, epoch derivation, ledger_urls). Verdict: DO NOT — the {75.16} consumption contract is exactly the kind of hard-won behaviour a swap would re-litigate.

Option 3 — Pin bump 1.0.3 → 1.0.7 (enabler, not a refactor)

Buys: SQL-injection column hardening, NUL-stripping in text/jsonb writes (a free robustness fix for LLM-extracted content), better error context, stats_group/ auto_refresh (likely unusable under the burn-guard architecture), halfvec. Effort M: test_coco_api_facade.py version-pin gate update, _coco_api source re-verification (the façade exists for exactly this), re-run of the {75.8}/{75.16}/{75.17} engine probes (all keepable tests exist), full pytest + staging smoke + LMDB semantics re-check (1.0.4 shared memo cache vs the burn runbook’s clear step). Risk MEDIUM — bounded by the façade + keepable probes, but a memo-format change across the bump can force a full reprocess anyway. Value: MEDIUM (security/robustness fixes; staying 4 releases behind has a rising cost). Verdict: worth a dedicated small Task, independent of any flow.py restructuring; do NOT couple it to Option 1.

Option 4 — Full cocoindex-native rewrite

Rebuild flow.py “the way the docs do it” (dataclass rows, from_class, vector index declaration, live mode, native IDs, exception handlers everywhere). NO-GO. Every §3.2 gap still binds (the observability/rollup/retry/ordering substrate must be rebuilt anyway), four §6 contracts would be violated outright (C-3 burn guard, C-8 PK scheme, DDL-via-CLI, FK model), and the memo/component-identity reset guarantees a full-corpus reprocess. Effort XL, value negative.

---

6. Engine-contract constraints — what any refactor MUST preserve

These are live-proven contracts. Violating any one reproduces a defect that has already cost real sessions.

• C-1 — {75.16} mount_each consumption contract. Any source with a watch attribute routes through the LiveMapFeed protocol: items are consumed EXCLUSIVELY via watch(subscriber); __aiter__ is reached only through subscriber.update_all(). watch() MUST call update_all() BEFORE mark_ready() (mark_ready-only feeds zero items, zero errors — the S319 {url: 0} defect). Executable record: scripts/tests/test_url_source_engine_consumption.py (4-case probe matrix).
• C-2 — {75.17} FK-tolerant deferred-flush convergence. mount_table_target rows flush AFTER component return; an in-component write referencing a declared-but- unflushed row MUST tolerate exactly its named FK violation (constraint_name string compare — never asyncpg class identity) and converge on walk 2. A raise instead routes into BI-19 containment, which DISCARDS the item’s declared rows → zero convergence forever.
• C-3 — Burn model (write-model §6 + bl-221/ID-83). Boot is lifespan-only (coco.start_blocking(), no app_main); ingestion fires ONLY on bearer-gated POST /walk → one-shot update_blocking(live=False). No refactor may introduce live mode, auto_refresh, or any boot-time walk.
• C-4 — {75.8} memo-over-frozen-dataclass. The engine fingerprints ("dataclass", module, qualname, field-values). Frozen dataclass memo keys are SUPPORTED; corollary: module/qualname are part of the fingerprint — moving or renaming memoised components or their argument dataclasses invalidates memo state.
• C-5 — {75.9} epoch-keyed memo. _pullmd_fetch(url, content_epoch) — the epoch is the re-fetch token; a URL-only memo silently serves stale markdown forever. Per-walk flow_op_id in ingest_url’s kwargs deliberately busts its memo each walk ({75.17} relies on the re-run); _pullmd_fetch’s own memo keeps that cheap. Do not “optimise” either key.
• C-6 — Write model R1–R5 (docs/themes/canonical-pipeline/reference/cocoindex-write-model.md): no cross-target FKs on pipeline-written tables (integrity via uuid5-by-construction); jsonb via the pool codec (never per-site json.dumps); natural-key dedup before declare_row (ON CONFLICT covers the PK only); FileLike.size()/content_fingerprint() are async and file_path.path is ABSOLUTE in production; the workspace manifest is walked as an item and must be skipped.
• C-7 — {66.19}/{66.16} dispatch boundary. Per-item components run on the _LoopRunner daemon thread: no ContextVar propagation (explicit kwargs via NAMED closures; re-bind locally), no functools.partial (engine requires __name__/__qualname__), component_subpath pinned to the historical identities ("ingest_file"/"ingest_url") — changing subpaths re-keys component state.
• C-8 — uuid5 PK scheme. uuid5(_KH_PIPELINE_DOC_NS, "<kind>:"+seed) is the idempotency AND integrity substrate (post-FK-drop) AND the TS-side join contract. Frozen.
• C-9 — {80.13} memo-HIT shape. A memo HIT returns a plain dict; every typed consumer normalises via a coercer before attribute access (else reconciliation GCs the prior walk’s rows, S316).
• C-10 — Collection-safety. No eager cocoindex.ops.* imports at module top; the _coco_api façade + function-local imports preserve bare-MagicMock test collection. Any restructuring keeps the façade as the single off-surface import point.
• C-11 — Observability contracts. The 7-key stage_counts map, 6-class error vocabulary (TS-mirrored, Zod-enforced), webhook payload shape, and PII redaction (bl-165 Option D) are externally consumed — refactors relocate but never reshape them.

---

7. NO-GO criteria + ratification options table

Explicit NO-GO criteria (any one stops the corresponding refactor)

1. Memo/identity reset without a budgeted window. Any change that moves/renames a @coco.fn(memo=True) component, its argument dataclass, or a component_subpath without a scheduled LMDB-clear + full-reprocess window and Anthropic burn sign-off.
2. Burn-guard violation. Anything that reintroduces boot-time walking, live mode, or auto_refresh (C-3).
3. PK re-keying. Adoption of resources.id generators or any change to the uuid5 scheme (C-8) — re-keys the corpus and breaks TS-side joins.
4. Out-of-band DDL. declare_vector_index / declare_sql_command_attachment / ManagedBy.SYSTEM adoption (DDL-via-CLI rule).
5. Contract regression on probes. Any slice whose pre-implementation probe contradicts C-1/C-2/C-9 semantics (e.g. coco.exception_handler proves observe-only, or reconciliation does not cover the trim’s shrink case) — that slice reverts to NO-GO, recorded {56.18}-style.
6. Capability-gap denial. Any spec that assumes per-row callbacks, retry kwargs, cross-target transactions, or ContextVar propagation — ABSENT in 1.0.3 and changelog-confirmed still absent through 1.0.7.

Options table (ratification-ready)

#	Option	Effort	Risk	Value	Recommendation
0	Do nothing	—	none	none	Acceptable default; revisit at next pipeline Task
1	Mechanical decomposition (non-memoised substrate only: observability + schemas + URL helpers to satellites; memoised components stay in flow.py)	S–M	MEDIUM → LOW after the memo-fingerprint probe	Comprehension + merge surface	RECOMMENDED, gated on the C-4 cross-module fingerprint probe
2a	TableSchema.from_class + dataclass rows	S–M	MEDIUM (encoders, OMITted columns)	Type-safety	Probe OMIT semantics first; spec only if clean
2b	coco.exception_handler containment	S	MEDIUM-HIGH (containment semantics unknown; BI-19 discard)	~40 lines	Probe-then-likely-skip
2c	Engine reconciliation replaces _trim_stale_form_fields	XS	LOW-MEDIUM	−30 lines, one fewer SQL seam	Best probe-to-value; run the shrink probe
2d	coco.map chunk fan-out	XS	LOW	Latency	Take opportunistically with any chunking work
2e	PgTableSource replaces FeedUrlSource internals	M	HIGH	~0	DO NOT (re-litigates {75.16})
3	Pin bump 1.0.3 → 1.0.7	M	MEDIUM (façade + keepable probes bound it)	Security/NUL fixes, currency	Separate small Task; decouple from Option 1
4	Full cocoindex-native rewrite	XL	CRITICAL	negative	NO-GO

---

8. Pre-ratification empirical verification (OQ-3 / Q-EX2)

• Date: 07/06/2026. Pinned version: cocoindex[postgres]==1.0.3 (requirements.txt:49); runtime cocoindex.__version__ == "1.0.3".
• Method: import-and-call probes (importlib + hasattr + inspect.signature) against the installed package, run in this session.

Symbol / surface	Result
coco.fn (sig: memo, memo_key, batching, max_batch_size, runner, version, logic_tracking='full', deps)	PRESENT — and confirms NO retry kwargs
coco.App / AppConfig / lifespan / mount / mount_each / use_mount / mount_target / map / component_subpath / ContextKey / use_context / runtime / start / GPU	PRESENT
coco.exception_handler (scoped, (handler) -> AsyncIterator[None]) + EnvironmentBuilder.set_exception_handler	PRESENT
connectors.postgres.mount_table_target (sig: (db, table_name, table_schema, *, pg_schema_name=None, managed_by=ManagedBy.SYSTEM)) — NO transaction/ordering/callback params	PRESENT
TableSchema.from_class; TableTarget.{declare_row, declare_vector_index, declare_sql_command_attachment}	PRESENT
connectors.postgres._source.PgTableSource; connectors.localfs.walk_dir; connectorkits.target.ManagedBy	PRESENT
ops.text.{RecursiveSplitter, SeparatorSplitter, detect_code_language}; ops.litellm.LiteLLMEmbedder	PRESENT
ops.entity_resolution.resolve_entities (sig: (entities, *, embedder, resolve_pair, is_existing_canonical=None, existing_policy=PINNED, on_resolution=None, max_distance=0.3, top_n=5))	PRESENT
resources.file.{FileLike, PatternFilePathMatcher}; resources.id.{generate_id, generate_uuid, IdGenerator}; resources.schema.VectorSchema; resources.chunk.Chunk	PRESENT
_internal.live_component.{LiveMapView, LiveMapFeed}	PRESENT
connectors.kafka	ABSENT on this install (optional extra cocoindex[kafka] not installed — expected, not a drift)
cocoindex.ExtractByLlm / LlmSpec / LlmApiType	ABSENT — not re-probed in this session; canonical record is knowledge-hub-archive/audits/cocoindex-1.0.3-extractbyllm-spec-reality-investigation.md. Checker audit confirmed ABSENT at runtime (07/06/2026)
Upstream latest	1.0.7 (PyPI JSON, 07/06/2026); 1.0.4–1.0.7 changelog reviewed (§3.3)

No SIGNATURE_DRIFT or BEHAVIOUR_DRIFT found between this document’s claims and the installed pin. Engine-behaviour claims (C-1/C-2/C-4/C-9) cite their existing keepable probe tests / live-evidence journals rather than fresh probes — the proposed NEW probes (cross-module memo fingerprint, exception-handler containment, reconciliation shrink) are explicitly listed as ratification-gated next steps, not assumed results.

---

9. Open questions for Liam (ratification gate)

1. Is Option 1 worth a Task at all right now? It is comfort-grade (navigation + merge surface), competes with v1-completion work, and carries the C-4 probe prerequisite. A legitimate answer is “park until the next time a Task must open flow.py anyway”.
2. Probe budget: approve the three XS probes (cross-module memo fingerprint; exception-handler containment semantics; reconciliation shrink case) as a single keepable-test spike subtask ({56.18} pattern), independent of any refactor decision?
3. Pin bump (Option 3): schedule as its own small Task on the backlog, or defer until upstream ships something KH actively needs? (The NUL-stripping fix is the strongest current pull.)
4. Scope guard: confirm Options 2e and 4 are recorded as NO-GO / DO-NOT-BUILD so future sessions do not re-derive them.

---

End {89.1} RESEARCH. Verdict: flow.py is cocoindex-native where the engine has a surface, and hand-rolled precisely where 1.0.3 (and, changelog-verified, 1.0.7) has none. The refactor opportunity is real but small, mechanical, and memo-fingerprint- gated — not the wholesale “we rebuilt what the library gives us for free” the S321 suspicion anticipated. No code, migrations, or ledger writes were made.