T8 TECH — Cocoindex flow scaffolding + Cloud Run sidecar deploy

Spec slug: cocoindex-flow-scaffolding (single spec — RATIFIED-S252 Wave 1) Subtask: ID-28.3 (TECH, succeeds {28.2 PRODUCT}, precedes {28.4 PLAN} per Q-PLANNER-2 fresh-per-Subtask discipline) Parent Task: ID-28 — T8 cocoindex flow scaffolding + Cloud Run sidecar deploy (docs/specs/id-31-canonical-pipeline-implementation-plan/PLAN.md §4.8) Critical-path position: Wave 4. Gates: T1 (Q-EX2 contract ratified) + T2 (Q-OQR1-16 combined-PR migration SHIPPED-S247) + T6 (q_a_extractions schema SHIPPED-S250 migration 20260520225456_t6_q_a_pairs_full_schema.sql) + T3 (RLS-pattern apply SHIPPED-S239). Gates T7 (Phew Q&A first-ingest) per RATIFIED-S243. Author: task-planner (opus-4-7, isolation=worktree) — FRESH INSTANCE per Q-PLANNER-2 (NOT the {28.1} RESEARCH author, NOT the {28.2} PRODUCT author).

§1. Context

T8 stands up two coupled implementation slices:

Cocoindex pipeline scaffolding at scripts/cocoindex_pipeline/flow.py — the 6-stage topology (source walk → binary conversion → LLM extraction → embedding → entity resolution → Postgres UPSERT) per docs/plans/phase-0-investigation/architecture/02-data-flow.md §3.1, hosted in a Cloud Run Service with min_instances=1, max_instances=1 per S14 single-orchestrator-instance topology.
Cloud Run sidecar deploy — a new Service alongside the existing per-tenant Jobs at cloudrun/jobs/{prod,staging}-{kpf,phew}.yaml. The Service hosts cocoindex + Docling (1.8 GB MIT) inside one container image (~5.3 GB target); pullmd remains a separately-deployed Service per O-Q3 (AGPL network-service boundary). Existing .github/workflows/cloud-run-deploy.yml is extended with a new Service-deploy step alongside the existing Jobs-deploy step.

What this TECH spec does:

Maps every PRODUCT invariant Inv-1..Inv-27 to one or more Proposed changes (P-1..P-9) with file paths, function signatures, schema diffs, migration shape, and acceptance-test shape.
Documents the v1 substrate for the 4 P-OQs surfaced by PRODUCT (audit_log v1 RAISE LOG only per P-OQ1; cocoindex defaults for retry per P-OQ2; cocoindex LMDB DLQ + structured logs + pipeline_runs.status='failed' for dead-letter per P-OQ3; both 35-file-corpus + per-file-p95 latency framings per P-OQ4).
Cross-links to the ratified Q-EX2 extraction contract at docs/specs/id-36-cocoindex-extraction-contract/TECH.md §3 (Path A canonical pattern — @coco.fn-wrapped direct anthropic call; the cocoindex ExtractByLlm primitive is ABSENT in 1.0.3 per S256) — does NOT restate Pydantic class shapes here.

What this TECH spec does NOT do (per PRODUCT §3 out-of-scope):

Author Pydantic ExtractionOutput class declarations — already ratified at docs/specs/id-36-cocoindex-extraction-contract/{PRODUCT,TECH}.md.
Introduce a new KH-owned dead-letter table — P-OQ3 ratified pipeline_runs.status='failed' rollup + cocoindex internal LMDB DLQ + structured logs. No new KH table.
Populate an audit_log table at v1 — P-OQ1 ratified: defer audit_log substrate to v1.1; v1 uses RAISE LOG per RLS-PATTERN P-5.
Author CocoInsight on-prem deployment topology — DEFERRED-v1.1 per O-Q7.
Stage Phew Q&A files into the source-binding folder — T7 owns that per O-Q8; T8 ships empty source-binding (Service running, fs-watch armed; T7 stages files post-T8 stable).

Source-of-truth documents (read in full to validate Proposed changes):

docs/specs/id-28-cocoindex-flow-scaffolding/RESEARCH.md — ratified S252 Wave 1.
docs/specs/id-28-cocoindex-flow-scaffolding/PRODUCT.md — ratified S252 Wave 2 (27 invariants + 4 P-OQ defaults accepted).
docs/specs/id-31-canonical-pipeline-implementation-plan/PLAN.md §4.8 — T8 detail.
docs/specs/id-36-cocoindex-extraction-contract/TECH.md §3 — Path A extraction pattern (@coco.fn direct anthropic call; cross-link target).
docs/plans/phase-0-investigation/architecture/02-data-flow.md §3-§5 + §7.2 — 6-stage topology, op_id hybrid, recordPipelineRun rollup.
docs/plans/phase-0-investigation/architecture/03-tech-stack.md §5-§7 — cocoindex + Docling + pullmd substrate detail.
lib/pipeline/record-run.ts — existing recordPipelineRun() helper signature (pipelineName, status, itemsProcessed, itemsCreated, workspaceId, sourceFilename, cost, progress, result, errorMessage, skipSentryAlert).
spike/cocoindex_s1/probe_managed_by_user.py — canonical live-wiring shape (coco.AppConfig(name=, main_fn=) + mount_table_target(managed_by=ManagedBy.USER)).
scripts/ontology-sync/parse-flow.py — canonical layered-fn-shape stub (outer process_ontology_file(file) + inner parse_cv_frontmatter(content_text: str)).
.github/workflows/cloud-run-deploy.yml — existing Cloud Run deploy baseline (WIF auth, per-tenant gcloud run jobs replace, Secret Manager mount).
cloudrun/cloudbuild.yaml — existing buildpack-based image build (~3.3 GB current).
cloudrun/jobs/{prod,staging}-{kpf,phew}.yaml — per-tenant Jobs manifest convention.
CLAUDE.md — cocoindex 1.0.3 sandbox gotcha; recordPipelineRun gotcha; content_text_hash GENERATED ALWAYS gotcha; anon-EXECUTE-via-PUBLIC gotcha; CLI .temp/project-ref drift gotcha.

§2. Proposed changes (P-1 through P-9)

Each Proposed change maps one or more PRODUCT invariants to a concrete implementation slice. Coverage matrix at §2.10 confirms 27/27 invariants covered.

P-1. Cloud Run sidecar Service deploy (Cocoindex + Docling co-located; pullmd separate)

Covers: Inv-6, Inv-7, Inv-8, Inv-9, Inv-10.

Implementation summary: Add a new Cloud Run Service (NOT Job) per O-Q2, alongside existing per-tenant Jobs. The Service hosts cocoindex engine (Rust + LMDB) + Docling (1.8 GB MIT) in one container image. pullmd remains a separately-deployed Service (existing self-hosted Docker stack per 03-tech-stack.md §7.3) — the cocoindex sidecar calls pullmd via HTTP per O-Q3 AGPL boundary.

LMDB cold-start strategy (O-Q6 ratified): LMDB ops-DB is ephemeral per container instance — no Cloud Storage persistent volume mount. On cold-start, cocoindex re-fingerprints the source corpus (~7 s for 35-file v1 canonical scale per Spike #2 measurement). Re-evaluate if corpus crosses 10 k files per v1.1 guidance — at that scale, Cloud Storage persistent-volume mount becomes the better trade-off vs re-fingerprint latency.

Sidecar version identification (Inv-8): The Cloud Build pipeline injects IMAGE_SHA (resolved git commit SHA at build time) as a Cloud Run Service environment variable; cocoindex sidecar reads os.environ["IMAGE_SHA"] at boot and passes it through to every recordPipelineRun() call via the result.extractor_version field. Forensic correlation: query pipeline_runs.result -> 'extractor_version' for any successful run; cross-reference the SHA against the Cloud Run deploy log (gcloud run services describe kh-cocoindex-pipeline-{tenant} --format='value(spec.template.spec.containers[0].env)').

Files touched / created:

cloudrun/services/{prod,staging}-{kpf,phew}-cocoindex.yaml — NEW. 4 per-tenant Service manifests (parallel-named with existing Jobs). Each declares min_instances=1, max_instances=1 per S14 + cpu=2, memory=4Gi (~5.3 GB image needs headroom; 2 Gi was the existing Jobs default, bumped 2x for Docling model in memory) + containerConcurrency=10 (cocoindex single-writer LMDB; allows in-process concurrency only) + serviceAccountName: {phew,kpf}-pipeline-sa@... (REUSE existing per-tenant SAs).
cloudrun/cloudbuild-cocoindex.yaml — NEW. Second cloudbuild config building a new image kh-cocoindex-pipeline (separate from existing kh-pipeline). Buildpack gcr.io/buildpacks/python with extra Docling pre-warm step that downloads layout-heron + docling-models at build time (NOT runtime) so cold-start does not block on model fetch per O-Q4. Image-size budget: ~5.3 GB target (3.3 GB existing baseline + 1.8 GB Docling + ~150 MB cocoindex transitive). Cloud Build 16 GB cap leaves headroom.
cloudrun/cocoindex-prewarm.py — NEW. Build-time script invoked by buildpack: imports Docling and triggers model download (docling.document_converter.DocumentConverter() lazy-loads on first call; pre-warm forces the fetch into the image layer). Runs under dangerouslyDisableSandbox: true per CLAUDE.md cocoindex gotcha.
.github/workflows/cloud-run-deploy.yml — EDIT. Add new job step Build cocoindex image via Cloud Build (parallel to existing Build images via Cloud Build step, using --config=cloudrun/cloudbuild-cocoindex.yaml). Add new step Deploy cocoindex Services (parallel to existing Deploy tenant manifests step, using gcloud run services replace cloudrun/services/${ENV_PREFIX}-${TENANT}-cocoindex.yaml). Reuse the same WIF auth, deploy SA, and Secret Manager mount (--set-secrets) step — extend the secrets loop to include both Job and Service deploys. Path-trigger guard extended: cloudrun/services/** added to paths: allow-list.
requirements.txt — EDIT. Append new pip pins:
- cocoindex[postgres]==1.0.3 (the version validated by S1 / S2 / S14 spikes).
- docling>=2.0.0,<3.0.0 (cap minor-version range; pinned exact at first deploy per O-Q4 cold-start tolerance).
- litellm>=1.40.0,<2.0.0 (per O-Q5 ratified default; verify Anthropic + OpenAI passthrough works under WIF SA at first deploy).
- asyncpg>=0.30.0 (cocoindex transitive; explicit-pin for reproducibility — the spike harness uses it directly).
- faiss-cpu==<pin> (REQUIRED by Stage-5 cocoindex.ops.entity_resolution.resolve_entities per OQ-C IN SCOPE — RATIFIED S265; CPU-only build for the Cloud Run sidecar; pin exact at first deploy; landed by subtask ID-28.29).
cloudrun/services/README.md — NEW. Operator-facing doc: Service vs Job split rationale (S14 + O-Q2), per-tenant SA mapping, smoke-verify command (gcloud run services describe kh-cocoindex-pipeline-{phew,kpf}), pre-warm-vs-runtime-download trade-off.

Migration shape: None. (Cloud Run topology change is infrastructure-only; no DDL.)

Acceptance test shape (per PRODUCT inv):

Inv-6: HTTPS health probe to https://<service-url>/health returns 200 OK from Vercel host within Cloud Run warm-start window. Integration test polls the staging Service URL after deploy step succeeds.
Inv-7: Ingest one file of each MIME (PDF, DOCX, XLSX, HTML, markdown) end-to-end via the canonical pipeline; each lands a content_items row with non-empty content_text column. Integration test inside __tests__/integration/cocoindex/sidecar-mime-coverage.test.ts.
Inv-8: Query pipeline_runs.result -> 'extractor_version' for any successful sidecar run; assert the field is present and non-empty, then cross-reference against the Cloud Run deploy log via gcloud run services describe kh-cocoindex-pipeline-{tenant} --format='value(spec.template.spec.containers[0].env)' to confirm SHA match. Integration test inside __tests__/integration/cocoindex/sidecar-version-metadata.integration.test.ts.
Inv-9: docker image inspect europe-west2-docker.pkg.dev/<project>/pipeline/kh-cocoindex-pipeline:latest returns zero pullmd binary entries + zero Playwright dependencies. Post-build assertion step in cloudbuild-cocoindex.yaml.
Inv-10: After Cloud Run Service scale-to-zero + cold-start cycle, first PDF ingest completes within 60 s end-to-end. Integration test injects scale-to-zero via gcloud run services update --min-instances=0 then immediately re-sets to 1 + measures first-call latency.

Effort estimate: ~6 h (3 h manifests + cloudbuild + workflow edits; 2 h pre-warm script + image-size verification; 1 h smoke + cold-start integration test wiring).

P-2. Cocoindex flow scaffolding at `scripts/cocoindex_pipeline/flow.py`

Covers: Inv-1, Inv-2, Inv-3, Inv-5.

Implementation summary: Author the canonical 6-stage cocoindex pipeline per 02-data-flow.md §3.1 on the installed 1.0.3 reactive App / Component / Target-reconciliation API (RESEARCH.md §R0–§R3). Module exports app_main() async function + KH_PIPELINE_APP = coco.App(coco.AppConfig(name="kh_pipeline"), app_main). app_main() (a) await mount_table_target(...) for each of the three target tables (returns three TableTarget handles); (b) opens the source feed via localfs.walk_dir(source_path, live=True, recursive=True).items() (per CLAUDE.md recursive=True gotcha + RESEARCH §1.2 — walk_dir returns a DirWalker, .items() yields (relative_path, File)); (c) fans out one component PER source item via await coco.mount_each(ingest_file, source.items(), ci, qa, sd). The six stages run INSIDE the per-item @coco.fn ingest_file(file, ci, qa, sd) — read → binary-convert → extract → embed → build row dict → declare_row(row={...}). The DB pool binds env-scope via a @coco.lifespan builder using builder.provide(DB_CTX, pool) (NOT the 2-arg use_context(DB_CTX, pool) async-CM form, which is wrong-arity in 1.0.3 — use_context(key) is single-arg read-only per RESEARCH §R1.7). The dataflow API (coco.Flow, flow[...], DataSlice.transform, flow.bind_target, collect/export) is ABSENT in 1.0.3 and is NOT used.

Structural-rewrite flag (RESEARCH §R6). The pre-S265 §P-2 sketch was built on the fictional dataflow API end-to-end (coco.Flow, flow["source"].transform(...), flow.bind_target(...)). The fix is a STRUCTURAL rewrite of app_main(), not a line-level bind_target → declare_row substitution: the transforms move from flow-scope .transform() chaining INTO the per-item ingest_file body, and DB-pool provisioning moves from use_context to @coco.lifespan. App boot (scripts/cocoindex_pipeline/__main__.py / any server.py) MUST register the lifespan on the environment — owned by ID-28.23. The substrate that SURVIVES unchanged: the three TableSchema declarations, _build_dsn(), the flow_context.py contextvars (bind_flow_meta / current_flow_meta / bind_retry_counter), and the rollup/error helpers — only their call-sites/wiring change.

Files touched / created:

scripts/cocoindex_pipeline/__init__.py — NEW. Empty package marker (Python package convention).

scripts/cocoindex_pipeline/flow.py — NEW. Module with:

1
#
2
# Cocoindex 1.0.3 canonical 6-stage pipeline (T8) — REACTIVE App / Component /
3
# Target-reconciliation API per RESEARCH.md §R0–§R3 (live PROBE PASS §R3).
4
# NOT the dataflow API: coco.Flow / flow[...] / .transform / flow.bind_target /
5
# collect / export are ABSENT in 1.0.3 and are NOT used.
6
#
7
# Topology (all six stages run INSIDE the per-source-item @coco.fn ingest_file):
8
#   1. source walk        -> connectors.localfs.walk_dir(path, live=True,
9
#                            recursive=True).items()  -> (rel_path, File) feed
10
#   2. binary conversion  -> per-MIME adapters (P-3): docling for PDF/DOCX/XLSX,
11
#                            pullmd HTTP client for HTML, passthrough for markdown
12
#   3. LLM extraction     -> Path A: KH @coco.fn extractors call anthropic SDK
13
#                            directly (extraction.py), validate via pydantic
14
#                            TypeAdapter — per cocoindex-extraction-contract TECH §3.1
15
#                            + docs/research/cocoindex-1.0.3-extractbyllm-spec-
16
#                            reality-investigation.md (ExtractByLlm ABSENT in 1.x)
17
#   4. embedding          -> KH-owned LiteLLMEmbedder call (OQ-B, ID-49.2; vector(1024)),
18
#                            written into content_items.embedding via declare_row.
19
#                            S274-AMENDED: `declare_vector_index` IS NOT CALLED —
20
#                            cocoindex 1.0.3 + 1.0.6 `_VectorIndexHandler` ignores
21
#                            `managed_by=USER` and would DROP/CREATE the index every
22
#                            cycle. The HNSW cosine index (m=16, ef_construction=64)
23
#                            is owned by migration `pre_squash_reconciliation.sql:4683`.
24
#                            See top-of-file S274 Last-verified note.
25
#   5. entity resolution  -> IN SCOPE v1 (OQ-C OVERRIDDEN by Liam, S265;
26
#                            subtask 28.29). resolve_entities() dedups/
27
#                            canonicalises across the run's entity_mentions
28
#                            (Stage 3). Needs faiss PINNED in requirements.txt.
29
#   6. Postgres UPSERT     -> TableTarget.declare_row(row={...}) per table; targets
30
#                            mounted via mount_table_target(managed_by=ManagedBy.USER)
31
#
32
# Fan-out: coco.mount_each(ingest_file, source.items(), ci, qa, sd) mounts one
33
# component per source item and calls ingest_file(File, ci, qa, sd) per item
34
# (RESEARCH §R1.11). NB coco.map() is pure concurrent execution with NO components
35
# — wrong primitive for the write path.
36
#
37
# DB pool: env-scope via @coco.lifespan + builder.provide(DB_CTX, pool). The 2-arg
38
# use_context(key, value) async-CM form is WRONG-ARITY in 1.0.3 — use_context(key)
39
# is single-arg read-only (RESEARCH §R1.7).
40
#
41
# Source-binding folder: env var COCOINDEX_SOURCE_PATH (T8 ships EMPTY default;
42
# T7 stages files post-T8 stable per O-Q8).
43
#
44
# Latency budgets (per PRODUCT inv-2 + P-OQ4):
45
#   - 35-file canonical corpus end-to-end ≤120 s (S2 cold-cache baseline)
46
#   - per-file p95 ≤30 s
47
#
48
# CLAUDE.md gotchas applied:
49
#   - localfs.walk_dir(recursive=True)  (default is False)
50
#   - cocoindex requires dangerouslyDisableSandbox in dev
51
#   - content_items.content_text_hash is GENERATED ALWAYS — omit from TableSchema
52
#   - recordPipelineRun() integration via per-flow op_id sidecar webhook (P-7)
53

54
from __future__ import annotations
55

56
import asyncio
57
import logging
58
import os
59
import uuid
60
from pathlib import Path
61

62
import asyncpg
63
import cocoindex as coco
64
from cocoindex.connectors import localfs
65
from cocoindex.connectors.postgres import (
66
    ColumnDef,
67
    TableSchema,
68
    TableTarget,
69
    mount_table_target,
70
)
71
from cocoindex.connectorkits.target import ManagedBy
72
# Stage 5 — native entity resolution (OQ-C IN SCOPE, S265; needs faiss pinned).
73
from cocoindex.ops.entity_resolution import resolve_entities
74

75
from scripts.cocoindex_pipeline.adapters import (
76
    convert_binary_to_markdown,  # P-3 outer-tier adapter (@coco.fn)
77
)
78
from scripts.cocoindex_pipeline.extraction import (
79
    # Path A — KH @coco.fn extractors call the anthropic SDK directly
80
    # (NOT cocoindex.functions.ExtractByLlm, which is ABSENT in 1.0.3).
81
    extract_classification,
82
    extract_entity_mentions,
83
    extract_qa_form,
84
)
85
from scripts.cocoindex_pipeline.flow_context import (
86
    bind_flow_meta,
87
    bind_retry_counter,
88
    current_flow_meta,  # reads op_id from FLOW_META_CTX inside ingest_file
89
)
90

91
ANTHROPIC_MODEL = "claude-opus-4-6"  # production tier per cocoindex-extraction-contract TECH §3.1
92
# Identity handle for the DB pool — provided env-scope via @coco.lifespan below.
93
DB_CTX = coco.ContextKey[asyncpg.Pool]("kh_pipeline_db")
94

95
CONTENT_ITEMS_SCHEMA = TableSchema(
96
    columns={
97
        "id": ColumnDef(type="uuid", nullable=False),
98
        "content_text": ColumnDef(type="text", nullable=True),
99
        "embedding": ColumnDef(type="vector(1024)", nullable=True),
100
        "op_id": ColumnDef(type="uuid", nullable=True),  # P-4: stamp per-flow op_id
101
        "source_document_id": ColumnDef(type="uuid", nullable=True),
102
        # content_text_hash GENERATED ALWAYS — OMITTED per CLAUDE.md gotcha
103
    },
104
    primary_key=("id",),
105
)
106

107
Q_A_EXTRACTIONS_SCHEMA = TableSchema(
108
    columns={
109
        "id": ColumnDef(type="uuid", nullable=False),
110
        "source_content_item_id": ColumnDef(type="uuid", nullable=True),
111
        "extractor_kind": ColumnDef(type="text", nullable=False),
112
        "extracted_question_text": ColumnDef(type="text", nullable=False),
113
        "extracted_answer_text": ColumnDef(type="text", nullable=True),
114
        "extraction_metadata": ColumnDef(type="jsonb", nullable=False),
115
        "op_id": ColumnDef(type="uuid", nullable=True),  # P-4
116
    },
117
    primary_key=("id",),
118
)
119

120
SOURCE_DOCUMENTS_SCHEMA = TableSchema(
121
    columns={
122
        "id": ColumnDef(type="uuid", nullable=False),
123
        "storage_path": ColumnDef(type="text", nullable=False),
124
        "content_fingerprint": ColumnDef(type="text", nullable=True),
125
        "op_id": ColumnDef(type="uuid", nullable=True),  # P-4 (requires migration P-7.M1)
126
    },
127
    primary_key=("id",),
128
)
129

130

131
@coco.fn(memo=True)
132
async def ingest_file(
133
    file: localfs.File,
134
    ci: TableTarget,
135
    qa: TableTarget,
136
    sd: TableTarget,
137
) -> None:
138
    """Per-source-item component (one per file via mount_each).
139

140
    memo=True ⇒ cocoindex skips re-invocation when the source bytes are
141
    unchanged, so a no-op re-ingest does NOT re-stamp op_id (PRODUCT Inv-11,
142
    OQ-A; RESEARCH §R4). All six stages run here; the write is declare_row.
143
    """
144
    # ── Stage 2: binary conversion (per-MIME adapter; @coco.fn(memo=True)) ──
145
    content_text = await convert_binary_to_markdown(file)  # P-3 adapter
146

147
    # ── Stage 3: LLM extraction — Path A, plain awaits over local content_text ─
148
    # NOT flow["content_text"].transform(...); extractors consume the LOCAL
149
    # `content_text` variable and call the anthropic SDK directly (extraction.py).
150
    classification = await extract_classification(content_text)
151
    qa_form = await extract_qa_form(content_text)
152
    entity_mentions = await extract_entity_mentions(content_text)
153

154
    # ── Stage 4: embedding — KH-owned embedder call (OQ-B), vector(1024) ────
155
    embedding = await embed_content(content_text)  # KH embedder; NOT ops.litellm
156

157
    # ── Stage 5: entity resolution (OQ-C IN SCOPE, S265; faiss-backed) ──────
158
    # dedup/canonicalise across this run's mentions → canonical entities written
159
    # to the entity tables. resolve_entities is cocoindex.ops.entity_resolution.
160
    resolved_entities = await resolve_entities(entity_mentions)
161

162
    # ── op_id — a PLAIN ROW FIELD read from FLOW_META_CTX (NOT bind_target) ──
163
    meta = current_flow_meta()  # flow_context.py contextvar; survives re-grounding
164
    sd_id = uuid.uuid4()
165
    ci_id = uuid.uuid4()
166

167
    # ── Stage 6: Postgres UPSERT — declare_row per table (op_id is a field) ──
168
    sd.declare_row(row={
169
        "id": sd_id,
170
        "storage_path": file.file_path,  # File exposes .file_path (NOT .path)
171
        "content_fingerprint": file.content_fingerprint,
172
        "op_id": meta.op_id,
173
    })
174
    ci.declare_row(row={
175
        "id": ci_id,
176
        "content_text": content_text,
177
        "embedding": embedding,
178
        "source_document_id": sd_id,
179
        "op_id": meta.op_id,
180
    })
181
    for mention in resolved_entities:  # canonicalised (Stage 5), not raw mentions
182
        qa.declare_row(row={
183
            "id": uuid.uuid4(),
184
            "source_content_item_id": ci_id,
185
            "extractor_kind": mention.extractor_kind,
186
            "extracted_question_text": mention.question_text,
187
            "extracted_answer_text": mention.answer_text,
188
            "extraction_metadata": mention.model_dump(),
189
            "op_id": meta.op_id,
190
        })
191
    # _emit_upsert_log() fires at THIS call-site per declare_row (P-5):
192
    # logs {op_id, table, row_id, operation:"upsert"} — "declared" not "applied",
193
    # no INSERT/UPDATE distinction (reconciler decides privately, RESEARCH §R8).
194

195

196
async def app_main() -> None:
197
    source_path = Path(os.environ.get("COCOINDEX_SOURCE_PATH", ""))
198
    if not source_path.exists():
199
        # T8 ships empty source-binding per O-Q8 — log + return (idle mode).
200
        # T7 stages files post-T8 stable, at which point COCOINDEX_SOURCE_PATH
201
        # is set in the Service manifest env block.
202
        logging.getLogger(__name__).info(
203
            "cocoindex source-binding folder missing or unset — App "
204
            "running in idle mode. Set COCOINDEX_SOURCE_PATH and restart "
205
            "when ready to stage files."
206
        )
207
        return
208

209
    # ── Stage 6 targets: mount three TableTargets (managed_by=USER; KH owns DDL) ─
210
    # mount_table_target is async (RESEARCH §R1.4); the DB pool is resolved from
211
    # DB_CTX, which the @coco.lifespan builder provided env-scope (see below).
212
    ci = await mount_table_target(
213
        DB_CTX, "content_items", CONTENT_ITEMS_SCHEMA, managed_by=ManagedBy.USER,
214
    )
215
    qa = await mount_table_target(
216
        DB_CTX, "q_a_extractions", Q_A_EXTRACTIONS_SCHEMA, managed_by=ManagedBy.USER,
217
    )
218
    sd = await mount_table_target(
219
        DB_CTX, "source_documents", SOURCE_DOCUMENTS_SCHEMA, managed_by=ManagedBy.USER,
220
    )
221
    # ── Stage 4: NO index hook here ───────────────────────────────────────────
222
    # S274-AMENDED: `ci.declare_vector_index(...)` IS NOT CALLED. cocoindex
223
    # 1.0.3 + 1.0.6 `_VectorIndexHandler._apply_actions` ignores
224
    # `managed_by=USER` and unconditionally DROP/CREATE INDEX every flow
225
    # update. The HNSW cosine index `idx_content_items_embedding`
226
    # (m=16, ef_construction=64) is owned by migration
227
    # `pre_squash_reconciliation.sql:4683`. Embedder writes the vector
228
    # column (Stage 4 inside ingest_file); migration owns the index.
229
    # Upgrade-watch: re-verify on every cocoindex upgrade. See top-of-file
230
    # S274 Last-verified note + `docs/research/id49-2-trade-off-analysis-s274.md`.
231

232
    # ── Stage 1: source walk (live fs-watch, nested recursive) → keyed feed ──
233
    source = localfs.walk_dir(source_path, live=True, recursive=True)
234

235
    # ── Fan-out: one component per source item; ingest_file(File, ci, qa, sd) ─
236
    run_op_id = uuid.uuid4()  # KH generates the run op_id (RESEARCH §R9 — cocoindex
237
                             # does NOT emit it; this is the faithful N7 realisation)
238
    async with bind_flow_meta(op_id=run_op_id), bind_retry_counter():
239
        await coco.mount_each(ingest_file, source.items(), ci, qa, sd)
240

241

242
def _build_dsn() -> str:
243
    """Build Postgres DSN from env vars (Cloud Run Secret Manager mount)."""
244
    # Mirrors spike/cocoindex_s1/probe_managed_by_user.py:build_dsn() shape.
245
    # SUPABASE_URL + SUPABASE_SERVICE_ROLE_KEY mounted via Cloud Run secrets
246
    # (P-1 sidecar manifest reuses the existing 11-secret mount loop).
247
    ...  # implementation detail; see spike harness for canonical shape
248

249

250
@coco.lifespan
251
def _kh_lifespan(builder: coco.EnvironmentBuilder):
252
    """Provide the asyncpg pool env-scope for the App's lifetime.
253

254
    Replaces the (wrong-arity) 2-arg `use_context(DB_CTX, pool)` async-CM. App
255
    boot (__main__.py / server.py) MUST register this lifespan on the
256
    environment — owned by ID-28.23. `builder.provide(DB_CTX, pool)` binds the
257
    handle; an async pool can instead use `builder.provide_async_with(...)`.
258
    """
259
    pool = asyncio.get_event_loop().run_until_complete(
260
        asyncpg.create_pool(_build_dsn(), min_size=2, max_size=10)
261
    )
262
    builder.provide(DB_CTX, pool)
263
    yield
264
    asyncio.get_event_loop().run_until_complete(pool.close())
265

266

267
# AppConfig has NO main_fn field — main_fn is the second positional to App
268
# (RESEARCH §R1.8).
269
KH_PIPELINE_APP = coco.App(coco.AppConfig(name="kh_pipeline"), app_main)
270

271

272
if __name__ == "__main__":
273
    # Reactive App entrypoint. The live probe (RESEARCH §R3) ran
274
    # `app.update_blocking(full_reprocess=..., live=...)`; the Service boot
275
    # (28.23) chooses the live/blocking mode + registers _kh_lifespan.
276
    KH_PIPELINE_APP.update_blocking(live=True)

embed_content (Stage 4, OQ-B). A KH-owned @coco.fn(memo=True) embedder that produces the vector(1024) written into content_items.embedding via declare_row. NOT cocoindex.ops.litellm.LiteLLMEmbedder (OQ-B): ops.litellm is not even importable on the installed pin (ModuleNotFoundError: No module named 'litellm' — RESEARCH §R1.10 / §8). text-embedding-3-large / vector(1024) per CLAUDE.md. Stage 4 is REQUIRED for vector search over the re-ingested corpus (a NULL-embedding corpus cannot serve search — RESEARCH §R5); recommend it lands as a dedicated subtask AFTER the bare write path is proven.

Stage 5 entity RESOLUTION — IN SCOPE v1 (RATIFIED — OQ-C OVERRIDDEN by Liam, S265). The re-ingested corpus ships with resolved/canonicalised entities at v1 (PRODUCT Inv-3). Proposed change: Stage 5 calls cocoindex.ops.entity_resolution.resolve_entities over the run’s entity mentions (Stage 3, Path A) to dedup/canonicalise across mentions, and the Stage-6 loop writes the resolved entities (see the ingest_file sketch above — resolved_entities = await resolve_entities(entity_mentions)). Dependency: resolve_entities needs the faiss package, which is NOT in requirements.txt today (empirically ModuleNotFoundError: No module named 'faiss' — RESEARCH §R1.10 / §8); faiss MUST be PINNED to a working version in requirements.txt (faiss-cpu for the CPU-only Cloud Run sidecar; pin exact at first deploy). This lands as a dedicated implementation subtask ID-28.29, sequenced AFTER the bare write path (28.21) is proven so the write path is de-risked first. RESEARCH §R5 records the OQ-C override.

scripts/cocoindex_pipeline/__main__.py — NEW. Module entrypoint so python3 -m scripts.cocoindex_pipeline boots the App (used by the Cloud Run Service GOOGLE_ENTRYPOINT). MUST register _kh_lifespan on the App environment before update_blocking() — owned by ID-28.23.

Migration shape: None for P-2 itself; P-4 lands the op_id ALTERs.

Acceptance test shape:

Inv-1: __tests__/integration/cocoindex/file-change-detection.integration.test.ts — write <source>/test.md; poll pipeline_runs table; assert exactly one new row with pipeline_name='kh_canonical_pipeline' appears within polling window (≤30 s).
Inv-2: __tests__/integration/cocoindex/latency-budget.integration.test.ts — write 35 files (canonical corpus shape from docs/client-documentation-corpus/); poll content_items for all 35 keys; assert total wall-clock ≤120 s AND per-file p95 ≤30 s (P-OQ4 dual budget).
Inv-3: __tests__/integration/cocoindex/stage-topology.integration.test.ts — inject a deliberate failure at the embedding stage (mock the KH embedder to raise); assert pipeline_runs.result.stage_failed='embedding'. Then inject failure at the LLM-extraction stage (mock the Path A anthropic.messages.create to raise); assert pipeline_runs.result.stage_failed='llm_extraction'.
Inv-5: __tests__/integration/cocoindex/nested-corpus.integration.test.ts — write a file at <source>/a/b/c/deep.md (3 levels deep); assert a content_items row is produced. Failure mode if recursive=True is silently dropped: file is never observed and assertion times out.

Effort estimate: ~5 h (3 h flow.py authoring + 1 h __main__.py + dsn helper + 1 h 4 integration tests).

P-3. Layered fn-shape per-MIME adapters in `scripts/cocoindex_pipeline/adapters.py`

Covers: Inv-4, Inv-5 (nested-corpus support flows through the adapter chain).

Implementation summary: Author the outer-tier @coco.fn(memo=True) source-binding adapter convert_binary_to_markdown(file: FileLike) -> str AND the inner-tier per-MIME extractor @coco.fn(memo=True) functions consuming content_text: str per S9 §7.2 layered fn-shape (COCO.10). The outer tier dispatches by MIME to Docling (PDF/DOCX/XLSX), pullmd HTTP client (HTML), or passthrough (markdown). The inner-tier extractors consume content_text: str so memoisation key is content bytes, not file handles — metadata-only edits (mtime, owner) do not re-trigger inner work.

Files touched / created:

scripts/cocoindex_pipeline/adapters.py — NEW. Module with:
- convert_binary_to_markdown(file: FileLike) -> str — outer-tier @coco.fn(memo=True). Routes by MIME: PDF/DOCX/XLSX → _docling_to_markdown(content_bytes: bytes) -> str (inner); HTML → _pullmd_to_markdown(url: str) -> str (inner, calls pullmd Service via HTTP); markdown → _passthrough_markdown(content_text: str) -> str (inner).
- _docling_to_markdown(content_bytes: bytes) -> str — inner @coco.fn(memo=True). Imports docling.document_converter.DocumentConverter (pre-warmed in image layer per P-1).
- _pullmd_to_markdown(url: str) -> str — inner @coco.fn(memo=True). HTTP POST to pullmd Service URL (PULLMD_SERVICE_URL env var, mounted via Cloud Run Secret Manager).
- _passthrough_markdown(content_text: str) -> str — inner @coco.fn(memo=True). Identity transform; memoisation buys idempotency on content-hash match.
Pattern mirrors scripts/ontology-sync/parse-flow.py outer/inner split. Inner-tier signatures take bytes (Docling) or str (pullmd URL, passthrough markdown) — NEVER FileLike per S9 §7.2.

Migration shape: None.

Acceptance test shape:

Inv-4: __tests__/integration/cocoindex/idempotency-memo-hit.integration.test.ts — ingest a file twice without modification; assert q_a_extractions row count for that source is identical pre/post second run AND the second run’s pipeline_runs.items_created is empty (memo-hit short-circuits all inner-tier work). Mirrors S2 spike Run B (warm-cache = 0 invocations).
Inv-5: Covered by P-2 acceptance test (the recursive adapter chain works through any depth).

Effort estimate: ~3 h (2 h adapter module + inner-tier delegations; 1 h memo-hit integration test).

P-4. `op_id` migration + per-row stamping at flow scope

Covers: Inv-11, Inv-12, Inv-14, Inv-15.

Implementation summary: Two coupled implementation steps:

Migration P-4.M1 — ALTER TABLE to add op_id uuid NULL to all three cocoindex target tables (content_items, q_a_extractions, source_documents) per O-Q1 ratification. This migration is UNCHANGED by the S265 re-grounding and remains correct — the columns are needed regardless of how the value is written. P-OQ1 follow-up: at v1 the per-table op_id stamping IS the audit-correlation substrate (Inv-12 round-trip); the audit_log.op_id column ratified by Inv-13 is DEFERRED-v1.1 per RLS-PATTERN P-5 (P-OQ1).
op_id as a declare_row row field in scripts/cocoindex_pipeline/flow.py — there is NO bind_target(op_id=) and NO flow["op_id"] in 1.0.3 (both ABSENT — RESEARCH §R1.1–§R1.2). KH generates run_op_id = uuid4() in app_main() and binds it into FLOW_META_CTX via bind_flow_meta(op_id=run_op_id); each per-item ingest_file reads it via current_flow_meta().op_id and includes it as a plain field of every declare_row(row={..., "op_id": meta.op_id}). Because the per-item fn is memo=True, an unchanged re-ingest is skipped and the row’s op_id is NOT re-stamped (PRODUCT Inv-11, OQ-A; RESEARCH §R4); full_reprocess=True re-stamps every row. flow_context.py (FLOW_META_CTX / bind_flow_meta / current_flow_meta) SURVIVES the re-grounding unchanged — only the CONSUMPTION site moves from the fictional bind_target(op_id=) to the row dict.

Files touched / created:

supabase/migrations/<timestamp>_t8_op_id_propagation.sql — NEW. DDL:

1
-- T8 W4 — op_id propagation column add per N7 hybrid pattern.
2
-- Spec: docs/specs/id-28-cocoindex-flow-scaffolding/TECH.md §P-4.
3
-- Ratification: O-Q1 (S252 W1) — T8 absorbs source_documents + q_a_extractions
4
-- ALTERs as T8-internal slice. PRODUCT Inv-11 / Inv-12 / Inv-14 / Inv-15.
5
--
6
-- Idempotency: IF NOT EXISTS guards on every ALTER so re-apply against any
7
-- env is a no-op.
8
--
9
-- Index rationale: GIN-free; op_id is a high-cardinality uuid, so a B-tree
10
-- index suffices for the Inv-12 round-trip query
11
-- (SELECT * FROM pipeline_runs WHERE op_id = <value>).
12
-- Pipeline_runs.op_id is similarly indexed below.
13

14
-- IMPORTANT: verify .temp/project-ref before push per CLAUDE.md gotcha.
15
-- IMPORTANT: no new PL/pgSQL functions in this migration so no anon-EXECUTE
16
-- REVOKE pattern needed; if downstream slices add stamping helpers,
17
-- those helpers MUST follow the canonical REVOKE FROM PUBLIC + REVOKE FROM anon
18
-- + GRANT TO authenticated, service_role pattern.
19

20
SET search_path = public, extensions;
21

22
-- 1. content_items.op_id (T2 combined-PR did NOT land this column — verified
23
--    via grep of supabase/migrations/20260520120828_t2_*.sql).
24
ALTER TABLE public.content_items
25
  ADD COLUMN IF NOT EXISTS op_id uuid NULL;
26
CREATE INDEX IF NOT EXISTS idx_content_items_op_id
27
  ON public.content_items (op_id) WHERE op_id IS NOT NULL;
28

29
-- 2. q_a_extractions.op_id (T6 migration 20260520225456_t6_* did NOT land
30
--    this column — verified).
31
ALTER TABLE public.q_a_extractions
32
  ADD COLUMN IF NOT EXISTS op_id uuid NULL;
33
CREATE INDEX IF NOT EXISTS idx_q_a_extractions_op_id
34
  ON public.q_a_extractions (op_id) WHERE op_id IS NOT NULL;
35

36
-- 3. source_documents.op_id (O-Q1 ratified T8 follow-up ALTER).
37
ALTER TABLE public.source_documents
38
  ADD COLUMN IF NOT EXISTS op_id uuid NULL;
39
CREATE INDEX IF NOT EXISTS idx_source_documents_op_id
40
  ON public.source_documents (op_id) WHERE op_id IS NOT NULL;
41

42
-- 4. pipeline_runs.op_id — REQUIRED for Inv-12 round-trip query
43
--    (op_id on row → SELECT pipeline_runs WHERE op_id = <value>).
44
--    pipeline_runs currently keys on id (uuid) + pipeline_name; op_id is
45
--    a NEW correlation key emitted by cocoindex per-flow.
46
ALTER TABLE public.pipeline_runs
47
  ADD COLUMN IF NOT EXISTS op_id uuid NULL;
48
CREATE INDEX IF NOT EXISTS idx_pipeline_runs_op_id
49
  ON public.pipeline_runs (op_id) WHERE op_id IS NOT NULL;
50

51
COMMENT ON COLUMN public.content_items.op_id IS
52
  'KH-generated per-run op_id, written as a declare_row field at UPSERT time per N7 hybrid (02-data-flow.md §5; cocoindex does NOT emit op_id — RESEARCH.md §R9). Round-trip: pipeline_runs.op_id.';
53
COMMENT ON COLUMN public.q_a_extractions.op_id IS
54
  'See content_items.op_id.';
55
COMMENT ON COLUMN public.source_documents.op_id IS
56
  'See content_items.op_id.';
57
COMMENT ON COLUMN public.pipeline_runs.op_id IS
58
  'Cocoindex per-flow op_id surfaced through recordPipelineRun(). Round-trip key for Inv-12.';

scripts/cocoindex_pipeline/flow.py — EDIT. Per the P-2 reactive shape: generate run_op_id in app_main(), bind it via bind_flow_meta(op_id=run_op_id), and include "op_id": current_flow_meta().op_id as a field of every declare_row(row={...}) inside ingest_file. No bind_target (ABSENT in 1.0.3).
lib/pipeline/record-run.ts — EDIT. Extend RecordPipelineRunParams interface to accept opId?: string | null (matches pipeline_runs.op_id uuid). Update the sb() insert payload to include op_id: opId ?? null.

Migration validation:

Pre-push: cat supabase/.temp/project-ref against expected value (CLAUDE.md gotcha — drift-after-env-flip).
Pre-push: column-drop-style audit not applicable (this is ADD COLUMN, not DROP) — no pg_trigger / pg_proc audit needed.
Apply via /opt/homebrew/bin/supabase db push with dangerouslyDisableSandbox: true.

Acceptance test shape:

Inv-11: __tests__/integration/cocoindex/op-id-stamping.integration.test.ts — ingest one file; query content_items, q_a_extractions, source_documents for the produced rows; assert all three op_id columns equal each other AND equal the pipeline_runs.op_id for the originating run.
Inv-12: Same test as Inv-11; extra assertion — given any pipeline-produced row’s op_id, SELECT * FROM pipeline_runs WHERE op_id = <value> returns exactly one row.
Inv-14: __tests__/integration/audit/non-pipeline-write.integration.test.ts — direct UI edit to a content_items row via service-role client; assert (P-5 deferred at v1, see open question OQ-T1 below) an audit_log row OR RAISE LOG line appears with NULL op_id. At v1 (P-OQ1 ratification: audit_log table population deferred), Inv-14 verification is via structured-log line inspection (Cloud Run log surface) rather than audit_log SELECT.
Inv-15: Re-ingest a file unchanged (memo-hit per Inv-4); assert NO new content_items.updated_at change AND NO new audit_log line (v1: no new structured-log line for no-op cycles).

Effort estimate: ~3 h (1.5 h migration + apply + 0.5 h record-run.ts edit + 1 h 4 integration tests).

P-5. Inv-13 v1 substrate — `RAISE LOG` only; `audit_log` table population DEFERRED-v1.1

Covers: Inv-13.

Implementation summary: Per P-OQ1 ratification (S252 W2 close), v1 does NOT populate an audit_log table for pipeline-driven writes. The v1 audit-observability path is structured logs (Cloud Run sidecar log ingest per RLS-PATTERN P-5 [DEFERRED-v1.1]). Inv-13 is retained as the behaviour-contract that v1.1 MUST satisfy when the audit_log table substrate lands.

Files touched / created:

None at v1 for the audit_log table itself.
scripts/cocoindex_pipeline/flow.py — EDIT (already in P-2). The App emits a structured log line per declare_row call via stdlib logging.getLogger(__name__).info(...) (coco.logger is absent in 1.0.3); the line carries op_id, table, row_id, operation: "upsert". S265 amend: the emission helper _emit_upsert_log() (landed at 28.10) is invoked at the declare_row call-site INSIDE ingest_file — this is the v1 path (NOT a Wave C TableTarget subclass). The line logs “declared” not “applied” and does not distinguish INSERT/UPDATE (RESEARCH §R8). Structured-log ingest by Cloud Run picks these up automatically; no KH-side log shipping plumbing needed beyond the existing Cloud Run log surface.
docs/specs/id-28-cocoindex-flow-scaffolding/TECH.md §4 (this spec) — flag Inv-13 v1.1 carry-forward as an open question OQ-T1 for Liam ratification.

Migration shape: None for v1.

Acceptance test shape (v1 only — v1.1 will replace with audit_log SELECT):

Inv-13 v1: __tests__/integration/cocoindex/audit-log-shipping.integration.test.ts — ingest one file; inspect the Cloud Run log stream (via gcloud logging read or the Cloud Run log emulator under test); assert at least one structured log line per Postgres UPSERT carrying op_id, table, row_id, operation. Test SKIPS if not running against a Cloud Run-shaped log surface; integration-only.

Effort estimate: ~1 h (0.5 h structured-log emission verification + 0.5 h integration test skeleton + skip-condition).

P-6. LLM extraction integration — Path A (`@coco.fn`-wrapped anthropic SDK)

Covers: Inv-20, Inv-21, Inv-22.

API-DEVIATION-S256 — Path A canonical pattern. cocoindex.functions.ExtractByLlm / cocoindex.LlmSpec / cocoindex.LlmApiType are ABSENT in the installed cocoindex==1.0.3 pin (empirically verified per OQ-3 import-and-call check at S256 W1; root-cause record in docs/research/cocoindex-1.0.3-extractbyllm-spec-reality-investigation.md). The pre-S256 P-6 cross-link assumed cocoindex’s ExtractByLlm primitive at flow scope; the actual canonical pattern in 1.x is a KH-authored @coco.fn(memo=True)-decorated function calling anthropic.AsyncAnthropic().messages.create(...) directly and validating the response via pydantic.TypeAdapter[...].validate_json(...). Verifier B-3 finding REVERSED: the pre-S256 anti-pattern call-out (“ExtractByLlm inside @coco.fn is API-incorrect”) no longer applies — in 1.x, @coco.fn IS the canonical extraction wrapper (confirmed by live examples-repo patient_intake_extraction_baml + patient_intake_extraction_dspy + paper_metadata). No litellm shim required — @coco.fn imposes zero LLM-provider gating.

Implementation summary: CROSS-LINK ONLY — DO NOT restate Q-EX2 Pydantic class shapes. The pipeline at scripts/cocoindex_pipeline/flow.py (per the P-2 reactive shape) imports the three Path A extractor functions (extract_classification, extract_qa_form, extract_entity_mentions) from scripts/cocoindex_pipeline/extraction.py — @coco.fn(memo=True)-decorated functions that internally hold the ExtractionOutput discriminated-union variants (ClassificationExtraction, QAFormExtraction, EntityMentionExtraction) and the three instruction prompts (CLASSIFICATION_PROMPT, Q_A_FORM_PROMPT, ENTITY_MENTION_PROMPT) authored by T1.2 (S256 W1 WP3 landed). S265 correction: the extractors are called as plain awaits INSIDE ingest_file(file, ...), consuming the LOCAL content_text variable produced by convert_binary_to_markdown(file) — NOT via flow["content_text"].transform(...) (coco.Flow / flow[...] / .transform are ABSENT in 1.0.3, RESEARCH §R1.1). The reactive model carries the per-source-item value as a local variable through the ingest_file body, so the extractor inputs are ordinary function arguments.

Path A invocation pattern: per docs/specs/id-36-cocoindex-extraction-contract/TECH.md §3.1 + docs/research/cocoindex-1.0.3-extractbyllm-spec-reality-investigation.md. The shape (extractor lives in extraction.py; called as a plain await in ingest_file):

1
# scripts/cocoindex_pipeline/extraction.py — Path A extractor (T1.2 module)
2
@coco.fn(memo=True)
3
async def extract_classification(content_text: str) -> ClassificationExtraction:
4
    client = anthropic.AsyncAnthropic()
5
    response = await _anthropic_retry(  # KH tenacity wrapper (§P-8); operative LLM retry
6
        client.messages.create,
7
        model="claude-opus-4-6",
8
        max_tokens=4096,
9
        messages=[{"role": "user", "content": f"{CLASSIFICATION_PROMPT}\n\n{content_text}"}],
10
    )
11
    return TypeAdapter(ClassificationExtraction).validate_json(response.content[0].text)
12

13
# scripts/cocoindex_pipeline/flow.py — consumed as a plain await inside ingest_file:
14
#   content_text = await convert_binary_to_markdown(file)
15
#   classification = await extract_classification(content_text)   # local var, not flow slice

Three such extractors (classification / q_a_form / entity_mentions). Memoisation per Inv-21 = memo=True + cocoindex content-hash determinism on the content_text argument + prompt version (when a prompt constant changes, the memo invalidates on the next run).

Validation-failure routing: per Q-EX2 Inv-13 (cocoindex-extraction-contract/PRODUCT.md), a Pydantic-validation failure produces a structured failure record in pipeline_runs with error_class, extraction_kind, content_items_id, raw-LLM-response-redacted. Since extraction is Path A (KH-issued, NOT a cocoindex-issued LLM call), the failure path is KH-side: TypeAdapter(...).validate_json(...) raises a pydantic.ValidationError inside ingest_file, which the per-stage exception handler classifies (error_class='extraction_validation_failed') and routes to _emit_pipeline_run_webhook(status='failed', ...) (P-7). There is NO cocoindex-internal DLQ in this chain — cocoindex’s native validation/DLQ machinery covers only cocoindex-issued calls (RESEARCH §R7). The pipeline_runs.status='failed' rollup (P-7) is the full path; no separate error-routing table.

Files touched / created:

scripts/cocoindex_pipeline/flow.py — EDIT (already in P-2). The three Path A extractor awaits inside ingest_file (consuming the local content_text) ARE the integration.
scripts/cocoindex_pipeline/extraction.py — the @coco.fn(memo=True) Path A extractors (T1.2 module; no new file from P-6 — cross-link only).

Migration shape: None.

Acceptance test shape:

Inv-20: __tests__/integration/cocoindex/extract-contract-honour.integration.test.ts — ingest one Q&A-form file + one classification-only file + one entity-mention file; assert the resulting q_a_extractions / content_items rows match the discriminator-keyed Pydantic shape from cocoindex-extraction-contract/PRODUCT.md Inv-1..Inv-14 (validation harness: Python pydantic.TypeAdapter[ExtractionOutput].validate_python(row.extraction_metadata)).
Inv-21: __tests__/integration/cocoindex/extract-memoisation.integration.test.ts — ingest one file; capture q_a_extractions rows; re-ingest same file unchanged; assert the new q_a_extractions rows are byte-identical to the first set (memo-hit determinism per S2 spike Run B).
Inv-22: __tests__/integration/cocoindex/extract-validation-failure.integration.test.ts — mock the Anthropic API to return malformed JSON (invalid discriminator); assert pipeline_runs row appears with status='failed', error_class='extraction_validation_failed', populated content_items_id; AND assert ZERO new rows in q_a_extractions / entity_mentions / content_items for that source.

Effort estimate: ~2 h (cross-link only; 0.5 h verifying import path + 1.5 h 3 integration tests).

P-7. `recordPipelineRun()` rollup integration

Covers: Inv-16, Inv-17, Inv-18, Inv-19.

Implementation summary: Bridge the cocoindex sidecar (Python) to the TS-side recordPipelineRun() helper (lib/pipeline/record-run.ts). Two viable architectures:

Option α (RECOMMENDED) — Sidecar webhook callback. The cocoindex sidecar emits a webhook to a new Vercel route /api/internal/pipeline-runs/record per pipeline invocation (start + end). The Vercel route calls recordPipelineRun() server-side. Authentication via shared secret + cron-secret-style header.

Option β — Python direct insert via supabase-py client. Rejected: bypasses the CLAUDE.md “Cron pipeline_runs inserts” gotcha (every cron-style write MUST go through recordPipelineRun() per Inv-18) and re-introduces Sentry-alerting + DB-error-routing inconsistencies that recordPipelineRun() exists to fix.

Files touched / created (Option α):

app/api/internal/pipeline-runs/record/route.ts — NEW. POST endpoint that:
- Authenticates via Authorization: Bearer <CRON_SECRET> header (matches existing cron-handler convention; CRON_SECRET already mounted via Secret Manager per P-1).
- Parses body: { opId: string, pipelineName: string, status: 'completed'|'completed_with_errors'|'failed', itemsProcessed: number, itemsCreated: string[], stageCounts: { source_walk: number, binary_conversion: number, llm_extraction: number, embedding: number, entity_resolution: number, postgres_upsert: number }, errorMessage?: string, errorClass?: string }.
- Calls recordPipelineRun() from @/lib/pipeline/record-run with pipelineName='kh_canonical_pipeline', all the body fields, AND the new opId field (per P-4 record-run.ts edit).
- Proxy gotcha: add /api/internal/pipeline-runs/record to publicRoutes allowlist in proxy.ts — but DON’T make it world-readable; auth gates via CRON_SECRET. Actually: cron-shape internal API routes DO go through proxy.ts but are excluded from auth-redirect at the per-route level since they handle their own auth. Verify against existing /api/cron/* pattern in proxy.ts.
scripts/cocoindex_pipeline/flow.py — EDIT. Add a flow-scope post-stage helper _emit_pipeline_run_webhook(op_id: uuid.UUID, status: str, stage_counts: dict, ...) invoked from app_main() at flow start (status='in_progress') + flow end (status='completed'|'completed_with_errors'|'failed'). HTTP POST via aiohttp to PIPELINE_RUN_WEBHOOK_URL env var (mounted via Secret Manager per P-1; points at https://kh.phew.org.uk/api/internal/pipeline-runs/record).
lib/pipeline/record-run.ts — EDIT (already in P-4 for opId). Add stageCounts?: Record<string, number> field to RecordPipelineRunParams so the per-stage counters land in pipeline_runs.result JSON column (existing field; not a new column).
__tests__/api/internal/pipeline-runs/record.test.ts — NEW. Unit test the new route’s auth + body parsing + recordPipelineRun call shape.

Migration shape: None. (Pipeline_runs op_id column already added in P-4.M1.)

Inv-19 verification (no pipeline_failures table):

__tests__/integration/pipeline/no-pipeline-failures-table.test.ts — runs SELECT to_regclass('public.pipeline_failures') against staging + prod; asserts result is NULL. Add to the always-on guard test suite.

Acceptance test shape:

Inv-16: Covered by P-2 acceptance (Inv-1 file-change-detection asserts exactly one pipeline_runs row per invocation). Extended: memo-hit polls (Inv-4 idempotency case) ALSO produce a pipeline_runs row (per PRODUCT Inv-16) — assert in __tests__/integration/cocoindex/memo-hit-pipeline-run.integration.test.ts: re-ingest a file twice; assert pipeline_runs count increments by 2, with the second row showing items_created=[] and result.stage_short_circuit='binary_conversion'.
Inv-17: P-2 stage-topology test extends — assert pipeline_runs.result.stage_counts exposes per-stage observability (e.g. stage_counts.binary_conversion=1, stage_counts.llm_extraction=1).

S273 ID-49.4 — stage-counter reality (layered observability). The §P-7 stage-counter aspiration is only PARTIALLY live. stage_counts["embedding"] is wired truthfully as of ID-49.4: a flow-scope _FlowStageCounter is bound via flow_context.bind_stage_counter around mount_each (mirroring the bind_retry_counter/_FlowRetryCounter pattern), ingest_file bumps "embedding" once per produced vector, and app_main folds the count into stage_counts["embedding"] before the flow-end webhook emit. The OTHER stage counters remain INERT in production: stage_counts["llm_extraction"] is only mutated by the _record_extraction_success helper, whose sole callers are unit tests (test_cocoindex_flow_stage_counts.py) — app_main never invokes it on the live per-item path because the reactive 1.0.3 model gives ingest_file no direct handle on app_main’s local stage_counts dict. Closing the remaining stages requires extending the same contextvar-binding substrate to each stage (or a per-item rollup channel) — flagged as an OQ / backlog candidate, cross-linked to backlog ID-158 (entity-resolution stage counts) and ID-162 (num_reprocesses observability). ID-49.4 deliberately bounds scope to the embedding counter (Inv-17 gap closure inherited from the ID-49.2 Checker) rather than rebuilding all of observability.

Inv-18: Code-discipline guard — __tests__/lib/pipeline/record-run-discipline.test.ts — runs bun scripts/ast-dataflow-cli.ts callers 'supabase.from(\\\'pipeline_runs\\\').insert' and asserts the result set is empty (no raw inserts in any pipeline code path).
Inv-19: to_regclass test described above.

Effort estimate: ~4 h (2 h new Vercel route + auth + body schema + 1 h Python webhook emitter + 1 h 4 integration / unit tests).

P-8. Failure-mode behaviour — retry / dead-letter / observability / no-silent-partial-writes

Covers: Inv-23, Inv-24, Inv-25, Inv-26, Inv-27.

Implementation summary: Apply the 3 P-OQ ratifications under the §R7 LAYERED retry reality (NOT a single cocoindex-native policy): P-OQ2 (RATIFIED OQ-E, S265) — KH-tenacity-owned LLM retry for Path A in extraction.py, cocoindex-native ONLY for cocoindex-issued LLM/embedding, ZERO per-row Postgres retry; P-OQ3 — pipeline_runs.status='failed' rollup + structured logs + cocoindex LMDB tracking (no per-row PG retry feeding it); P-OQ4 latency dual-budget. No new KH dead-letter table per P-OQ3 (aligns COCO.7 RATIFIED-DO-NOT-BUILD).

Files touched / created:

scripts/cocoindex_pipeline/extraction.py — the OPERATIVE LLM retry lives HERE, not in flow.py: the _anthropic_retry tenacity wrapper (@retry(stop=stop_after_attempt(3), wait=wait_exponential(...))) wraps each Path A anthropic.messages.create, with a before_sleep hook that bumps the flow-scope retry counter via bind_retry_counter() (RESEARCH §R7). This wrapper is real and works. RATIFIED (OQ-E, S265): 3 attempts / exponential backoff is the ratified v1 policy.
scripts/cocoindex_pipeline/flow.py — EDIT. Document the LAYERED retry policy in the module docstring per P-OQ2/OQ-E: (a) KH-tenacity for Path A LLM (operative, in extraction.py); (b) cocoindex-native ONLY for cocoindex-issued LLM/embedding (ops.litellm, unused by KH — and not even importable on the pin per §R1.10); (c) ZERO per-row Postgres retry — a transient PG failure surfaces as a component exception that cocoindex re-attempts on the next update cycle, not an in-run per-row retry.
scripts/cocoindex_pipeline/flow.py — EDIT. Augment _emit_pipeline_run_webhook() (P-7) to emit status='failed' + populated errorClass (per Inv-25 enum: extraction_validation_failed, extraction_provider_unavailable, postgres_write_failed, binary_conversion_failed, embedding_failed, entity_resolution_failed) when the layered retry exhausts (KH-tenacity attempts exhaust for LLM; first hard error for PG).
scripts/cocoindex_pipeline/flow.py — EDIT. Augment stdlib logging.getLogger(__name__).error(...) emissions (S254 amend — coco.logger absent in 1.0.3) at each per-stage exception handler to include {op_id, stage, error_class, content_items_id, redacted_error_message} JSON shape per Inv-26. PII redaction via existing patterns (the LLM-response payload is redacted before logging — only error_class + error_message_first_200_chars + op_id etc.).
lib/pipeline/error-classes.ts — NEW. Export the 6-class enum (extraction_validation_failed | extraction_provider_unavailable | postgres_write_failed | binary_conversion_failed | embedding_failed | entity_resolution_failed) as both a TS string-literal union AND a Zod enum for the app/api/internal/pipeline-runs/record/route.ts schema validation.

No-silent-partial-writes (Inv-27): No new code. Cocoindex’s postgres.mount_table_target() uses INSERT ... ON CONFLICT DO UPDATE per-row; PG transaction-per-row means a mid-pipeline failure rolls back the in-flight UPSERT but earlier-committed UPSERTs from the same run are persistent (NOT all-or-nothing across the flow). The Inv-27 guarantee is interpreted as “no PARTIAL ROW writes” (UPSERT is atomic per-row) — NOT “no partial RUN writes” (mid-run failure CAN leave content_items with 5/10 expected rows and q_a_extractions with 3/10 derived rows). This is the cocoindex-native behaviour and the integration test asserts it accordingly.

Migration shape: None.

Acceptance test shape:

Inv-23: __tests__/integration/cocoindex/transient-retry.integration.test.ts — mock the Path A anthropic.messages.create endpoint to 503 once then 200; assert the KH-tenacity wrapper retries, the pipeline run succeeds, AND pipeline_runs.result.retry_count=1 (the flow-scope counter bumped by the tenacity before_sleep hook).
Inv-24: __tests__/integration/cocoindex/persistent-failure-dlq.integration.test.ts — mock the Path A anthropic endpoint to 503 on every call; assert (a) pipeline_runs.status='failed' after KH-tenacity attempts exhaust, (b) the failure is enumerable via Cloud Run log SELECT (gcloud logging read 'jsonPayload.op_id=<value>'), (c) NO new KH dead-letter table SELECT needed.
Inv-25: Same test as Inv-24 — assert pipeline_runs.error_class='extraction_provider_unavailable' AND populated pipeline_runs.completed_at.
Inv-26: Same test as Inv-24 — assert at least one structured log line with required fields parseable as JSON.
Inv-27: __tests__/integration/cocoindex/no-partial-row-writes.integration.test.ts — mock the Postgres UPSERT to refuse on row 5 of a 10-row run (asyncpg.PostgresError); assert (a) rows 1-4 persist (per-row UPSERT atomicity is the v1 contract), (b) row 5 NOT in content_items, (c) pipeline_runs.status='failed' rolls up the run-level outcome. Test name + assertion comment clarify the per-row-vs-per-run semantics so the test is not misread later.

Effort estimate: ~4 h (1 h flow.py retry + log + error-class wiring + 0.5 h error-classes.ts + 2.5 h 5 integration tests).

P-9. Integration test suite design + corpus reuse

Covers: Cross-cutting — every PRODUCT invariant has at least one acceptance test referenced in P-1 through P-8.

Implementation summary: New test directory __tests__/integration/cocoindex/ with one test file per failure mode + happy path. Tests run via bun run test:integration (real Anthropic + Supabase per CLAUDE.md). Canonical test corpus: reuse docs/client-documentation-corpus/ (35 files validated by S2 spike). New fixture folder for failure-injection tests: __tests__/integration/cocoindex/fixtures/ with synthetic PDFs/DOCXs/HTML pages designed to trigger specific failure modes.

Files touched / created:

__tests__/integration/cocoindex/ — NEW directory.
__tests__/integration/cocoindex/fixtures/ — NEW directory with synthetic test files (~10 fixtures covering each MIME and failure mode).
__tests__/integration/cocoindex/test-helpers.ts — NEW. Shared helpers: pollContentItemsFor(key, timeoutMs), assertPipelineRunMetrics(opId, expectedStageCounts), injectLlmFailure(once: boolean), scaleSidecarToZero().
__tests__/integration/cocoindex/file-change-detection.integration.test.ts (Inv-1).
__tests__/integration/cocoindex/latency-budget.integration.test.ts (Inv-2).
__tests__/integration/cocoindex/stage-topology.integration.test.ts (Inv-3 + Inv-17).
__tests__/integration/cocoindex/idempotency-memo-hit.integration.test.ts (Inv-4).
__tests__/integration/cocoindex/nested-corpus.integration.test.ts (Inv-5).
__tests__/integration/cocoindex/sidecar-mime-coverage.integration.test.ts (Inv-7).
__tests__/integration/cocoindex/sidecar-version-metadata.integration.test.ts (Inv-8).
__tests__/integration/cocoindex/sidecar-pullmd-separation.test.ts (Inv-9; runs at build time as image-inspect step).
__tests__/integration/cocoindex/sidecar-cold-start.integration.test.ts (Inv-10).
__tests__/integration/cocoindex/op-id-stamping.integration.test.ts (Inv-11 + Inv-12).
__tests__/integration/audit/non-pipeline-write.integration.test.ts (Inv-14, v1 substrate via log inspection).
__tests__/integration/cocoindex/audit-log-shipping.integration.test.ts (Inv-13 v1).
__tests__/integration/cocoindex/memo-hit-pipeline-run.integration.test.ts (Inv-15 + Inv-16).
__tests__/lib/pipeline/record-run-discipline.test.ts (Inv-18 — unit-level grep guard).
__tests__/integration/pipeline/no-pipeline-failures-table.test.ts (Inv-19).
__tests__/integration/cocoindex/extract-contract-honour.integration.test.ts (Inv-20).
__tests__/integration/cocoindex/extract-memoisation.integration.test.ts (Inv-21).
__tests__/integration/cocoindex/extract-validation-failure.integration.test.ts (Inv-22).
__tests__/integration/cocoindex/transient-retry.integration.test.ts (Inv-23).
__tests__/integration/cocoindex/persistent-failure-dlq.integration.test.ts (Inv-24 + Inv-25 + Inv-26).
__tests__/integration/cocoindex/no-partial-row-writes.integration.test.ts (Inv-27).

Integration-test runtime gotchas:

bun run test:integration requires dangerouslyDisableSandbox: true for the cocoindex sub-process per CLAUDE.md.
Pinned test corpus path via COCOINDEX_SOURCE_PATH=<repo>/docs/client-documentation-corpus/ env var.
- Note: This <repo>/docs/client-documentation-corpus/ path is the integration-test corpus binding only. The production runtime COCOINDEX_SOURCE_PATH resolves to the B1 /corpus persistent disk (owned by ID-62.6 / ID-66.9 / ID-45), not a repo-relative path — keep the two distinct. The repo dir was renamed …-base/ → …-corpus/; the -base name is dead.
Each test sets up a unique sub-folder under the source path to avoid cross-test interference; teardown clears it.
Sidecar boot timeout: tests skip if kh-cocoindex-pipeline-{phew,kpf} Service is not reachable at staging URL (avoids local-dev false negatives).

Migration shape: None.

Acceptance test shape: This IS the acceptance test plan. Each test file owns one or more PRODUCT invariants; the coverage matrix at §2.10 maps every invariant to its test file.

Effort estimate: ~6 h (1 h helpers + fixtures setup + 5 h 22 integration test bodies + assertions).

§2.10 Coverage matrix (PRODUCT inv → Proposed change → acceptance test)

PRODUCT Inv	Area	Proposed change(s)	Acceptance test file
Inv-1	A	P-2	`file-change-detection.integration.test.ts`
Inv-2	A	P-2	`latency-budget.integration.test.ts`
Inv-3	A	P-2	`stage-topology.integration.test.ts`
Inv-4	A	P-3	`idempotency-memo-hit.integration.test.ts`
Inv-5	A	P-2 (covers P-3 chain)	`nested-corpus.integration.test.ts`
Inv-6	B	P-1	(workflow step + smoke deploy verify)
Inv-7	B	P-1	`sidecar-mime-coverage.integration.test.ts`
Inv-8	B	P-1	`sidecar-version-metadata.integration.test.ts`
Inv-9	B	P-1	`sidecar-pullmd-separation.test.ts`
Inv-10	B	P-1	`sidecar-cold-start.integration.test.ts`
Inv-11	C	P-4	`op-id-stamping.integration.test.ts`
Inv-12	C	P-4	`op-id-stamping.integration.test.ts` (extended)
Inv-13	C	P-5 (v1 substrate per P-OQ1)	`audit-log-shipping.integration.test.ts` (v1.1 substrate replaces)
Inv-14	C	P-4 (v1: log-inspection substrate)	`non-pipeline-write.integration.test.ts`
Inv-15	C	P-4	`memo-hit-pipeline-run.integration.test.ts`
Inv-16	D	P-7	`memo-hit-pipeline-run.integration.test.ts`
Inv-17	D	P-7	`stage-topology.integration.test.ts` (extended)
Inv-18	D	P-7 (code-discipline)	`record-run-discipline.test.ts`
Inv-19	D	P-7 (no-op guard)	`no-pipeline-failures-table.test.ts`
Inv-20	E	P-6 (cross-link to Q-EX2)	`extract-contract-honour.integration.test.ts`
Inv-21	E	P-6	`extract-memoisation.integration.test.ts`
Inv-22	E	P-6	`extract-validation-failure.integration.test.ts`
Inv-23	F	P-8 (P-OQ2 defaults)	`transient-retry.integration.test.ts`
Inv-24	F	P-8 (P-OQ3 substrate)	`persistent-failure-dlq.integration.test.ts`
Inv-25	F	P-8	`persistent-failure-dlq.integration.test.ts`
Inv-26	F	P-8	`persistent-failure-dlq.integration.test.ts`
Inv-27	F	P-8	`no-partial-row-writes.integration.test.ts`

Coverage: 27 / 27 PRODUCT invariants mapped to at least one Proposed change AND at least one acceptance test.

§3. Migration plan

ONE migration: supabase/migrations/<timestamp>_t8_op_id_propagation.sql per P-4. Adds op_id uuid NULL + B-tree partial index on content_items, q_a_extractions, source_documents, pipeline_runs. All ADDs are IF NOT EXISTS-guarded (idempotent re-apply against any env).

Apply procedure:

Pre-push verification (CLAUDE.md gotcha): cat supabase/.temp/project-ref — must match the expected project (turayklvaunphgbgscat for staging, rovrymhhffssilaftdwd for prod). Relink via supabase link --project-ref <correct> if drift detected.

Pre-push column-existence verification (defensive — guards against future T2/T6 follow-up landing op_id ahead of T8):

1
SELECT column_name FROM information_schema.columns
2
WHERE table_schema='public'
3
  AND column_name='op_id'
4
  AND table_name IN ('content_items','q_a_extractions','source_documents','pipeline_runs');

If any column already exists, the IF NOT EXISTS clause silently no-ops — safe.

Apply via CLI (NOT MCP apply_migration per CLAUDE.md): /opt/homebrew/bin/supabase db push with dangerouslyDisableSandbox: true (POSTGRES_PASSWORD env var set).
Post-apply verification: re-run the column query above; assert all 4 columns present.
Type regen: /opt/homebrew/bin/supabase gen types typescript --project-id rovrymhhffssilaftdwd --schema public > supabase/types/database.types.ts and commit.

Risk: None of the ALTERs is destructive (ADD COLUMN NULL has no rewrite cost on PG ≥ 11). Index creation uses default CREATE INDEX (non-CONCURRENTLY) — on the canonical-pipeline-implementation table sizes (~thousands of rows) this completes in <1 s. If prod scale exceeds 1M rows on any of the 4 target tables, swap to CREATE INDEX CONCURRENTLY and split into two migrations (ALTERs + indexes) — but at v1 scale this is overkill.

Why no new PL/pgSQL functions: P-4 introduces no helpers; cocoindex stamps op_id at the application layer (Python). No anon-EXECUTE-via-PUBLIC pattern needed. If a v1.1 follow-up adds e.g. audit_log_emit(op_id, table, row_id, op_type) helper, that migration MUST apply the canonical REVOKE EXECUTE ON FUNCTION public.foo(...) FROM PUBLIC; REVOKE EXECUTE ON FUNCTION public.foo(...) FROM anon; GRANT EXECUTE ON FUNCTION public.foo(...) TO authenticated, service_role; pattern.

§4. Open questions for Liam ratification before {28.4} PLAN dispatch

Each OQ here is a TECH-level uncertainty that surfaced during Proposed-change authoring. None blocks TECH ratification; all need a recommended default to be applied at PLAN dispatch (which can be amended by Liam in the ratification gate).

ID	OQ	Surfaced in	Recommended default for PLAN dispatch	Cost of deferral
T-OQ1	Inv-13 v1.1 follow-up Subtask creation — does T8 PLAN.md include a placeholder Subtask for the v1.1 `audit_log` table population work (substrate satisfying Inv-13 as ratified), OR is that work routed to a separate Task / backlog item created post-T8?	P-5 Inv-13 v1 deferral per P-OQ1.	Route to backlog as a new ID (post-T8 close). PLAN.md should NOT bloat with v1.1 follow-ups that depend on RLS-PATTERN v1.1 work. Create backlog item once T8 ships.	LOW — clean v1 scope; v1.1 follow-up enumerable from spec deferral notes.
T-OQ2	Webhook auth shape for the `/api/internal/pipeline-runs/record` route (P-7) — reuse `CRON_SECRET` (existing cron-handler convention; matches the Cloud Run Service’s secret mount) OR introduce a dedicated `PIPELINE_INTERNAL_SECRET`?	P-7.	Reuse `CRON_SECRET` — already mounted via Secret Manager loop in `cloud-run-deploy.yml`; one fewer secret to rotate. Document the reuse in the route handler’s header comment so future readers don’t mistake the shared-secret scope.	LOW — single-secret reuse is well-precedented in the cron-handler pattern; rotation cadence aligns.
T-OQ3	Per-tenant Service vs single-Service-multi-source-binding — does T8 ship 4 per-tenant Services (mirrors existing Jobs pattern: `{prod,staging}-{kpf,phew}-cocoindex`), OR 2 Services (one per env) with a single source-binding folder for ALL tenants?	P-1.	Per-tenant (4 Services). Matches existing per-tenant SA + Secret Manager pattern. Tenant isolation at the Cloud Run boundary is a load-bearing security property (data-residency + multi-tenant compromise blast-radius); collapsing to 2 Services breaks that.	LOW — Cloud Run Service per-tenant cost is small; isolation is worth more.
T-OQ4	Image-size budget enforcement — does the cloudbuild step ENFORCE the ~5.3 GB target (fail the build if exceeded) OR just log + warn? Cold-start latency hit per O-Q4 was the original budget motivator; hard enforce would catch dep bloat early.	P-1.	Log + warn at v1; ratchet to hard-enforce post-baseline measurement. First deploy is the baseline; if subsequent deploys grow by > 20% (e.g. Docling 2.x release with bigger models), the warn surfaces it. Hard-enforce risks blocking time-sensitive deploys on bloat that isn’t yet load-bearing.	LOW — soft start; tighten via a Subtask in T13 (operational gates Wave 6) if image bloat becomes a real issue.
T-OQ5	Inv-27 partial-write semantics — TECH §P-8 reads PRODUCT Inv-27 as “no PARTIAL ROW writes” (cocoindex UPSERT atomicity per-row) rather than “no PARTIAL RUN writes” (mid-run failure CAN leave rows 1-4 persisted with rows 5-10 missing). Is this the intended PRODUCT semantic, OR did Inv-27 mean run-level all-or-nothing (which would require KH-side flow-scope transaction wrapping)?	P-8 Inv-27 interpretation.	Per-row atomicity (the cocoindex-native semantic) — flow-scope transaction wrapping is anti-cocoindex (defeats incremental Δ pattern). If Liam intended run-level all-or-nothing, escalate as a new PRODUCT invariant; cocoindex’s architectural shape does NOT support it cheaply.	MEDIUM — semantic clarification only; integration test in P-9 documents the as-shipped behaviour either way.

§5. Effort estimate

Proposed change	Effort
P-1 Cloud Run sidecar Service deploy	~6 h
P-2 Cocoindex flow scaffolding (`flow.py`)	~5 h
P-3 Layered fn-shape per-MIME adapters	~3 h
P-4 op_id migration + per-row stamping	~3 h
P-5 Inv-13 v1 substrate (`RAISE LOG` only)	~1 h
P-6 ExtractByLlm integration (cross-link)	~2 h
P-7 `recordPipelineRun()` rollup integration	~4 h
P-8 Failure-mode behaviour	~4 h
P-9 Integration test suite design	~6 h
Total	~34 h

Note vs PLAN.md §4.8 estimate (“~4-5h spec authoring; impl deferred to S253+”): the PLAN.md estimate is for the spec-authoring portion (which has already completed across S252 W1+W2+W3 = the {28.1}+{28.2}+{28.3} planner dispatches). The ~34 h above is the IMPLEMENTATION effort that {28.4} PLAN decomposes into TM-shape Subtasks {28.5+} for S253+ dispatch. Maps roughly to 9-12 Subtasks well within the 25-Subtask soft ceiling per workflow-orchestration §3.4.

§6. Cross-references

§6.1 Source-of-truth spec lineage

Predecessor specs (this directory):
- docs/specs/id-28-cocoindex-flow-scaffolding/RESEARCH.md — ratified S252 W1 (RESEARCH.md §1-§5 + 8 OQ defaults).
- docs/specs/id-28-cocoindex-flow-scaffolding/PRODUCT.md — ratified S252 W2 (27 invariants + 4 P-OQ defaults).
Successor (fresh Planner per Q-PLANNER-2): docs/specs/id-28-cocoindex-flow-scaffolding/PLAN.md {28.4} — decomposes ratified PRODUCT + TECH pair into TM-shape implementation Subtasks {28.5+}.
Sibling specs (cross-linked, NOT restated):
- docs/specs/id-36-cocoindex-extraction-contract/{PRODUCT,TECH}.md — Q-EX2 ExtractByLlm contract (T1.1+T1.2 — Pydantic class shapes, validation behaviour, instruction prompts).
- docs/specs/id-36-cocoindex-ledger-api/TECH.md — TS-facing ledger API DEFERRED-v1.1; v1 substrate is pipeline_runs rollup (P-7).
- docs/specs/id-56-content-model-invariants/ — Q1.3-Q1.N content-model invariants (T1.4; consumed by cocoindex flow at extraction time).
- docs/specs/id-38-rls-pattern/{PRODUCT,TECH}.md — P-1..P-5 + T-1..T-5; P-5 observability [DEFERRED-v1.1] is the rationale for P-OQ1 audit_log table deferral.

§6.2 Architecture sub-doc sources

docs/plans/phase-0-investigation/architecture/02-data-flow.md §3 (6-stage topology) + §4 (Cloud Run sidecar) + §5 (op_id hybrid) + §7.2 (recordPipelineRun() rollup) + §10.4 (pipeline_failures DO-NOT-BUILD) + §10.5 (anti-patterns).
docs/plans/phase-0-investigation/architecture/03-tech-stack.md §5 (cocoindex recurring substrate) + §6 (Docling) + §7 (pullmd retention).
docs/plans/phase-0-investigation/0.9-decision-graph.md §11.4.1 (N7 op_id hybrid ratification).
docs/plans/phase-0-investigation/0.9-spike-S1-cocoindex-schema-coupling.md — CLOSED-S230 (Scenario A).
docs/plans/phase-0-investigation/0.9-spike-S2-cocoindex-folder-binding.md — CLOSED-S229 (memo-hit + localfs).
docs/plans/phase-0-investigation/0.9-spike-S14-cocoindex-concurrency.md — CLOSED-S230 (single-orchestrator topology).

§6.3 Canonical-pipeline + ops sources

docs/specs/id-31-canonical-pipeline-implementation-plan/PLAN.md §3 (dependency graph — T8 critical-path position) + §4.8 (T8 detail).
.github/workflows/cloud-run-deploy.yml — existing Cloud Run deploy baseline (WIF + per-tenant gcloud run jobs replace + Secret Manager mount loop).
cloudrun/cloudbuild.yaml — existing buildpack build (~3.3 GB current; P-1 extends with second config).
cloudrun/jobs/{prod,staging}-{kpf,phew}.yaml — per-tenant Jobs manifest convention (P-1 mirrors as Services).
lib/pipeline/record-run.ts — existing recordPipelineRun() helper (P-4 + P-7 extend with opId + stageCounts).
spike/cocoindex_s1/probe_managed_by_user.py — canonical live-wiring shape (P-2 mirrors).
scripts/ontology-sync/parse-flow.py — canonical layered fn-shape stub (P-3 mirrors).
docs/runbooks/cloud-run-phase-1.md — Cloud Run Phase 1 runbook (P-1 extends with sidecar-Service operator guidance).

§6.4 CLAUDE.md gotchas applied (per Proposed change)

P-1, P-2, P-3, P-9: cocoindex 1.0.3 requires dangerouslyDisableSandbox: true for both PyPI install and Rust-engine LMDB startup in dev. localfs.walk_dir defaults recursive=False — explicit recursive=True per P-2.
P-2, P-4: content_items.content_text_hash is GENERATED ALWAYS — omit from TableSchema declarations.
P-4: .temp/project-ref drift — pre-push verification per §3 migration plan. Migration uses IF NOT EXISTS for idempotency.
P-4 (v1.1 follow-up): if Inv-13 v1.1 substrate adds PL/pgSQL helpers, anon-EXECUTE-via-PUBLIC pattern MUST apply (both REVOKE FROM PUBLIC AND REVOKE FROM anon required; GRANT TO authenticated, service_role).
P-7: recordPipelineRun() from @/lib/pipeline/record-run — never raw insert (Inv-18 enforces).
P-7: new /api/internal/pipeline-runs/record route — verify against existing cron-handler proxy.ts pattern; do NOT silently redirect via publicRoutes allowlist omission.
P-7, P-8: silent-failure prevention — use sb() / tryQuery() from @/lib/supabase/safe (the recordPipelineRun() helper already does this; the new Vercel route inherits the pattern).
P-1 cloudbuild + P-9 integration tests: Bun fetch hangs on HTTP 204 through sandbox proxy — supabase-writing test helpers in __tests__/integration/cocoindex/ need dangerouslyDisableSandbox: true.

§7. Verification (S265 ID-28.20 amendment — OQ-3 empirical import-and-call check)

Per the OQ-3 pre-ratification forcing function, every external cocoindex symbol the S265 amendments cite (or remove) was re-checked against the installed pin via inspect.signature / hasattr / importlib.import_module, sandbox-disabled per the CLAUDE.md LMDB gotcha. The exhaustive log is RESEARCH.md §R1; this section records the subset load-bearing for the §P-2/§P-4/§P-5/§P-6/§P-8 corrections.

Date: 25/05/2026.
Pinned version: cocoindex[postgres]==1.0.3 (requirements.txt); installed cocoindex.__version__ == "1.0.3".

PRESENT (the reactive write path the amendments now cite):

Symbol	Result	Bears on
`cocoindex.App`, `AppConfig`	PRESENT (`AppConfig` has NO `main_fn` field — `main_fn` is the 2nd positional to `App`)	§P-2 `KH_PIPELINE_APP`
`cocoindex.mount_each`	PRESENT — mounts one component per `(key, value)`; calls `fn(value, *args)` per item	§P-2 fan-out
`cocoindex.connectors.postgres.mount_table_target(db, table_name, table_schema, *, pg_schema_name, managed_by=ManagedBy.SYSTEM)`	PRESENT, async	§P-2 three target handles
`TableTarget.declare_row(*, row: RowT) -> None`	PRESENT — keyword-only `row`	§P-2/§P-4 write path; §P-5 emission site
`TableTarget.declare_vector_index(*, column, metric, method, …)`	PRESENT	§P-2 Stage-4 pgvector index
`localfs.walk_dir(path, *, live=False, recursive=False, …) -> DirWalker`; `DirWalker.items()`	PRESENT (`recursive` defaults False — CLAUDE.md gotcha holds; `.items()` yields `(rel_path, File)`; `File.file_path` / `.read()` / `.content_fingerprint`)	§P-2 Stage-1 feed
`cocoindex.lifespan`, `EnvironmentBuilder.provide` / `provide_with` / `provide_async_with`	PRESENT	§P-2 DB-pool env-scope
`cocoindex.use_context(key) -> T`	PRESENT — single-arg, read-only	§P-2 (confirms 2-arg `use_context(DB_CTX, pool)` async-CM is WRONG-arity)
`ManagedBy.USER` (`= "user"`)	PRESENT — cocoindex writes rows, never DDL	§P-2 mounts
`cocoindex.ops.text.RecursiveSplitter`	PRESENT (native chunking, no extra dep)	(informational — available if KH chunks natively)

ABSENT (the dataflow API the amendments removed) — SIGNATURE_DRIFT-equivalent, resolved by removal:

Symbol	Result	Removed from
`coco.Flow` / `coco.flow` / `DataSlice` / `FlowBuilder` / `flow_def`	ABSENT	§P-2 (`coco.Flow` + `flow[...]` sketch)
`DataSlice.transform` (no DataSlice type exists)	ABSENT	§P-2/§P-6 (`flow["content_text"].transform(...)`)
`TableTarget.bind_target`	ABSENT	§P-2/§P-4 (`flow.bind_target(...)` + `op_id=flow["op_id"]`)
`coco.collect` / `coco.export` / `coco.add_collector`	ABSENT	(the 28.20-mandate’s own prescribed fix — RESEARCH §R0)
`cocoindex.functions.ExtractByLlm` / `coco.LlmSpec` / `coco.LlmApiType`	ABSENT (`ModuleNotFoundError`)	§P-6 (confirms S256 record; Path A stands)
`cocoindex.ops.litellm` (importable?)	ABSENT (`ModuleNotFoundError: No module named 'litellm'`)	§P-2/§P-8 (OQ-B: KH-owned embedder, NOT `ops.litellm`; reinforces KH-tenacity LLM retry)
`cocoindex.ops.entity_resolution` (importable?)	ABSENT (`ModuleNotFoundError: No module named 'faiss'`)	§P-2 Stage-5 (OQ-C OVERRIDDEN, S265 — entity RESOLUTION IS in scope v1; the missing-`faiss` finding is now the PIN REQUIREMENT, `faiss-cpu` to be pinned in `requirements.txt` per subtask 28.29 — NOT a deferral)

Escalation status: every ABSENT symbol was a fiction the amendment REMOVES (not a citation the spec depends on), so no spec citation is left dangling against the installed pin. The reactive replacements are all PRESENT with the signatures recorded above, and the end-to-end write path is PROBE-PASS-proven (RESEARCH §R3). No ABSENT/SIGNATURE_DRIFT finding blocks ratification.

End of TECH.md. S265 ratification complete: OQ-A + OQ-E RATIFIED (PROVISIONAL markers lifted); OQ-C OVERRIDDEN — Stage-5 entity resolution IS in scope at v1 (§P-2; faiss-cpu pinned; subtask ID-28.29). Re-grounded substrate consumed by the ID-28.21+ implementation Subtasks. The fictional dataflow API is removed; the reactive App/mount_each/declare_row path is the canonical contract per RESEARCH.md §R0–§R9.

T8 TECH — Cocoindex flow scaffolding + Cloud Run sidecar deploy

T8 TECH — Cocoindex flow scaffolding + Cloud Run sidecar deploy

§1. Context

§2. Proposed changes (P-1 through P-9)

P-1. Cloud Run sidecar Service deploy (Cocoindex + Docling co-located; pullmd separate)

P-2. Cocoindex flow scaffolding at scripts/cocoindex_pipeline/flow.py

P-3. Layered fn-shape per-MIME adapters in scripts/cocoindex_pipeline/adapters.py

P-4. op_id migration + per-row stamping at flow scope

P-5. Inv-13 v1 substrate — RAISE LOG only; audit_log table population DEFERRED-v1.1

P-6. LLM extraction integration — Path A (@coco.fn-wrapped anthropic SDK)

P-7. recordPipelineRun() rollup integration

P-8. Failure-mode behaviour — retry / dead-letter / observability / no-silent-partial-writes

P-9. Integration test suite design + corpus reuse

§2.10 Coverage matrix (PRODUCT inv → Proposed change → acceptance test)

§3. Migration plan

§4. Open questions for Liam ratification before {28.4} PLAN dispatch

§5. Effort estimate

§6. Cross-references

§6.1 Source-of-truth spec lineage

§6.2 Architecture sub-doc sources

§6.3 Canonical-pipeline + ops sources

§6.4 CLAUDE.md gotchas applied (per Proposed change)

§7. Verification (S265 ID-28.20 amendment — OQ-3 empirical import-and-call check)

P-2. Cocoindex flow scaffolding at `scripts/cocoindex_pipeline/flow.py`

P-3. Layered fn-shape per-MIME adapters in `scripts/cocoindex_pipeline/adapters.py`

P-4. `op_id` migration + per-row stamping at flow scope

P-5. Inv-13 v1 substrate — `RAISE LOG` only; `audit_log` table population DEFERRED-v1.1

P-6. LLM extraction integration — Path A (`@coco.fn`-wrapped anthropic SDK)

P-7. `recordPipelineRun()` rollup integration