coreci/inboxos-v1
18
0
Fork 0
Code Issues Pull Requests Actions Packages Projects 1 Releases 3 Wiki Activity

live: patch Papra_Link onto NocoDB rows after attachment write (REQ-INB-25)

Browse Source 
The historical pipeline captures Papra_Link during the pre-classify
attachment walk and writes it onto the NocoDB row at extract time.
The live pipeline previously emitted the papra_decision log event
but did not PATCH the link onto the row, creating a documented
asymmetry. This commit closes it.

pipelines/live_pipeline.py:
- Capture the first returned id from papra_write_attachment as
  papra_link (parallel to the historical pipeline's
  pre-classify walk).
- After the per-module upsert loop, iterate over the collected
  upserts and PATCH Papra_Link onto each module row that was
  upserted. Idempotent: a re-run with the same papra_link is
  a no-op on the column.
- Emit a structured papra_link_patched event with the message_id,
  the papra_link, and the list of modules_patched. Failures are
  per-row papra_link_patch_failed events; the pipeline continues.
- The papra_write_body branch also captures its single id into
  papra_link (defensive; body uploads go to one module, but
  the parity patch is harmless if there are 0 upserts).

docs/extraction-contract.md:
- §'Live vs historical: Papra_Link asymmetry' subsection removed.
  The asymmetry is closed; the cross-references to ARCHITECTURE.md
  and papra-write-policy.md are updated by T2.3.

ARCHITECTURE.md:
- §'Document Store (curated)' 'Live vs historical asymmetry'
  paragraph replaced with a single sentence: 'Both pipelines
  capture Papra_Link on the NocoDB row whenever the attachment
  walk returns a papra_link.' with a forward pointer to
  docs/extraction-contract.md.

pipelines/_tests/test_papra_link_parity_offline.py (new, ~1 test):
- test_papra_link_patched_on_live_after_attachment_write
  asserts the parity patch: 2 module rows + 2 Papra_Link patches,
  one papra_link_patched event listing both modules.

The papra_decide function in pipelines/common.py is unchanged;
the patch is a caller-side change in live_pipeline.py only.
docs/papra-write-policy.md §Rule 1 'Live pipeline note' removal
is deferred to T2.3 (Wave 2) per the plan.

---ci---
project: inboxos
phase: 1
milestone: v1.1
status: execute
---/ci---
This commit is contained in:
cloudinit-bot
2026-06-08 21:17:56 +00:00
parent 6067008b94
commit f44e2e2b4a
4 changed files with 308 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.ciagent/inboxos/ARCHITECTURE.md
+4 -8
View File
@@ -62,14 +62,10 @@
- **Papra** (`docker/docker-compose.yml`): Curated document store for selected attachments + bodies. **First ship is dry-run** — the pipeline emits Papra write intent but does not call the real Papra API. See `docs/papra-write-policy.md`.
**Live vs historical asymmetry.** The historical pipeline captures
`Papra_Link` during the pre-classify attachment walk (see
`pipelines/historical_pipeline.py:296-336`); the live pipeline calls
`papra_decide` once after classify and does not pre-walk
attachments. This asymmetry is intentional — see
`docs/extraction-contract.md §Live vs historical: Papra_Link
asymmetry` and `docs/papra-write-policy.md §Rule 1 — attachments
present`.
Both pipelines capture `Papra_Link` on the NocoDB row whenever the
attachment walk returns a `papra_link`; see
`docs/extraction-contract.md` and the `papra_link_patched` event in
`pipelines/live_pipeline.py`.
### Orchestration
docs/extraction-contract.md
-28
View File
@@ -152,34 +152,6 @@ version-controlled in Gitea, not embedded in n8n workflows"):
fails loudly if the first non-whitespace token is not valid JSON.
There is no "best effort" path.
## Live vs historical: Papra_Link asymmetry
The two pipelines have a documented asymmetry around `Papra_Link`:
- **Historical pipeline** (`pipelines/historical_pipeline.py:296-336`)
calls `papra_write_attachment` *before* the classify step during
the attachment walk. If any attachment is written, the first
document id is captured and passed to `extract_for_module` as
`papra_link`; the subsequent NocoDB row gains a `Papra_Link`
field.
- **Live pipeline** (`pipelines/live_pipeline.py:490-510`) calls
`papra_decide` once per message, *after* classify. Attachment
decisions are still recorded, but the pre-classify walk is
skipped — the live pipeline never pays the per-message cost of an
attachment pre-walk.
This is **intentional** in this phase. The historical pipeline
already pays the LLM round-trip cost for the backfill batch, and
capturing `Papra_Link` during the walk is essentially free. The
live pipeline optimises for the 5-minute cron, where one round
trip per message is the dominant cost. The structured log
(`papra_decision` event) is the per-message source of truth for
which documents would have been written in a live Papra run.
See `docs/papra-write-policy.md §Rule 1 — attachments present`
("Live pipeline note:") for the policy view, and
`ARCHITECTURE.md §Document Store (curated)` for the system view.
## Legacy: opencode CLI backend
`INBOXOS_LLM_BACKEND=opencode` is **kept for offline tests only**.
pipelines/_tests/test_papra_link_parity_offline.py
+260
View File
@@ -0,0 +1,260 @@
"""Offline smoke for the live-pipeline Papra_Link parity (REQ-INB-25).
Asserts that when the live pipeline's `process_one_message` returns
from a successful `papra_write_attachment` call with a non-None
`papra_link`, every module row that was upserted in this run has
`Papra_Link` set to the returned value. The asymmetry between
live and historical pipelines documented in
`docs/extraction-contract.md §Live vs historical: Papra_Link
asymmetry` is closed by this code path.
The test stubs the network layers (`gitea_request`, `requests.*`,
`papra_decide`, `nocodb_upsert`) and feeds a single in-memory
fixture message through `process_one_message`, then inspects the
NocoDB fake to assert the parity patch landed.
Run from the inboxos/ project root:
python3 pipelines/_tests/test_papra_link_parity_offline.py
"""
from __future__ import annotations
import base64
import email
import email.policy
import json
import logging
import os
import sys
from pathlib import Path
from typing import Any
from unittest.mock import patch
HERE = Path(__file__).resolve().parent
PIPE_DIR = HERE.parent
sys.path.insert(0, str(PIPE_DIR))
# ── Fixtures ─────────────────────────────────────────────────
PDF_PAYLOAD = b"%PDF-1.4\n%minimal\n"
MESSAGE_WITH_PDF = (
b"Message-ID: <pdf-parity@example.com>\r\n"
b'From: "Vendor" <v@vendor.com>\r\n'
b"Subject: Your invoice\r\n"
b"Date: Thu, 01 May 2026 12:00:00 +0000\r\n"
b"MIME-Version: 1.0\r\n"
b'Content-Type: multipart/mixed; boundary="BND"\r\n'
b"\r\n"
b"--BND\r\n"
b"Content-Type: text/plain; charset=utf-8\r\n"
b"\r\n"
b"Invoice attached.\r\n"
b"--BND\r\n"
b'Content-Type: application/pdf; name="invoice.pdf"\r\n'
b'Content-Disposition: attachment; filename="invoice.pdf"\r\n'
b"Content-Transfer-Encoding: base64\r\n"
b"\r\n"
)
MESSAGE_WITH_PDF += base64.b64encode(PDF_PAYLOAD) + b"\r\n--BND--\r\n"
def _silent_logger() -> logging.Logger:
lg = logging.getLogger("inboxos-papra-link-smoke")
lg.addHandler(logging.NullHandler())
lg.propagate = False
return lg
# ── Fakes ────────────────────────────────────────────────────
class _FakeNoco:
"""Records every upsert call so the test can inspect the rows."""
def __init__(self) -> None:
self.calls: list[dict[str, Any]] = []
# message_id -> {table_id: {row_id: record}}
self.rows: dict[str, dict[str, dict[int, dict[str, Any]]]] = {}
def upsert(self, cfg, table_id, record, message_id):
self.calls.append({
"table_id": table_id,
"message_id": message_id,
"record": dict(record),
})
if message_id not in self.rows:
self.rows[message_id] = {}
if table_id not in self.rows[message_id]:
self.rows[message_id][table_id] = {}
existing = self.rows[message_id][table_id]
if existing:
row_id = next(iter(existing))
existing[row_id].update(record)
return {"action": "patch", "id": row_id, "record": existing[row_id]}
row_id = len(existing) + 1
existing[row_id] = dict(record)
return {"action": "post", "id": row_id, "record": existing[row_id]}
# ── Test ─────────────────────────────────────────────────────
def test_papra_link_patched_on_live_after_attachment_write() -> None:
"""Live pipeline patches Papra_Link onto every module row when an attachment is written."""
from common import LLMConfig, NocoConfig, PapraConfig
import live_pipeline
msg = email.message_from_bytes(MESSAGE_WITH_PDF, policy=email.policy.default)
# NocoConfig / LLMConfig / PapraConfig (all dry-run for the Papra side)
nocodb_cfg = NocoConfig(
base_url="https://noco.example.test",
api_token="tok",
project_id="p",
)
llm_cfg = LLMConfig(
backend="ollama",
prompt_dir="./.opencode/commands",
opencode_bin="opencode",
ollama_base_url="https://ollama.example.test",
ollama_api_key=None,
default_model="gemma4:e4b",
extract_model="gemma4:e4b",
extract_command_prefix="inboxos:extract-",
timeout=60,
max_tokens=4096,
)
papra_cfg = PapraConfig(
base_url="https://papra.example.test",
api_token="tok",
organization_id="org",
body_write_modules=["crm", "finance", "legal"],
metadata_only_modules=["intel", "promotions"],
dry_run=True,
dry_run_override=None,
timeout=60,
)
# Module tables: one finance + one crm.
module_tables = {"finance": "tbl_finance_x", "crm": "tbl_crm_y"}
fake_noco = _FakeNoco()
# The live pipeline's classify + extract path is also stubbed:
# we want the upsert loop to write 2 rows and then the Papra_Link
# patch to write 2 more (one PATCH per row).
def fake_classify(*args, **kwargs):
return ["finance", "crm"]
def fake_extract(llm_cfg, module, summary, logger):
return {"Message_ID": summary["message_id"], "Some_Col": f"value-{module}"}
logger = _silent_logger()
captured: list[dict[str, Any]] = []
def fake_papra_write_attachment(cfg, decision, *, message_id, module, logger):
# Pretend Papra accepted the attachment and returned a doc id.
return ["doc-parity-12345"]
def fake_papra_decide(*, message_id, modules, msg, cfg):
from common import PapraDecision, PapraTarget, _content_hash
# Build one attachment target.
target = PapraTarget(
kind="attachment",
filename="invoice.pdf",
content_type="application/pdf",
payload=PDF_PAYLOAD,
content_hash=_content_hash(PDF_PAYLOAD),
)
return PapraDecision(
would_write=True,
reason="attachment_present",
module=None,
targets=[target],
)
def fake_save_state(s3, bucket, state):
# No-op: avoid SeaweedFS writes in the offline test.
pass
def fake_s3_put_bytes(s3, bucket, key, body, content_type="application/octet-stream"):
pass
# s3 client stub: only the methods the live pipeline uses.
class _FakeS3:
pass
# Patch the live pipeline's symbol references.
with patch.object(live_pipeline, "papra_decide", side_effect=fake_papra_decide), \
patch.object(live_pipeline, "papra_write_attachment", side_effect=fake_papra_write_attachment), \
patch.object(live_pipeline, "classify", side_effect=fake_classify), \
patch.object(live_pipeline, "extract_for_module", side_effect=fake_extract), \
patch.object(live_pipeline, "nocodb_upsert", side_effect=fake_noco.upsert), \
patch.object(live_pipeline, "s3_put_bytes", side_effect=fake_s3_put_bytes), \
patch.object(live_pipeline, "save_state", side_effect=fake_save_state), \
patch.object(live_pipeline, "log_event",
side_effect=lambda logger, event, **fields: captured.append({"event": event, **fields})):
ok = live_pipeline.process_one_message(
uid=1,
raw=MESSAGE_WITH_PDF,
s3=_FakeS3(),
s3_bucket="inboxos",
nocodb_cfg=nocodb_cfg,
llm_cfg=llm_cfg,
module_tables=module_tables,
papra_cfg=papra_cfg,
logger=logger,
)
assert ok is True, "process_one_message returned False"
# We expect 4 nocodb_upsert calls: 2 module rows + 2 Papra_Link patches.
assert len(fake_noco.calls) == 4, (
f"expected 4 upsert calls (2 rows + 2 patches), got {len(fake_noco.calls)}"
)
# The last 2 calls must be the Papra_Link patches.
patch_calls = [c for c in fake_noco.calls if c["record"].get("Papra_Link")]
assert len(patch_calls) == 2, (
f"expected 2 Papra_Link patch calls, got {len(patch_calls)}"
)
patched_table_ids = {c["table_id"] for c in patch_calls}
assert patched_table_ids == {"tbl_finance_x", "tbl_crm_y"}, (
f"expected patches on both tables, got {patched_table_ids}"
)
for c in patch_calls:
assert c["record"]["Papra_Link"] == "doc-parity-12345"
# The papra_link_patched event must have been emitted.
parity_events = [e for e in captured if e.get("event") == "papra_link_patched"]
assert len(parity_events) == 1, (
f"expected 1 papra_link_patched event, got {len(parity_events)}"
)
ev = parity_events[0]
assert ev["papra_link"] == "doc-parity-12345"
assert sorted(ev["modules_patched"]) == ["crm", "finance"]
# ── Runner ──────────────────────────────────────────────────
def _run() -> int:
tests = [
test_papra_link_patched_on_live_after_attachment_write,
]
failures = []
for t in tests:
try:
t()
print(f"PASS {t.__name__}")
except Exception as exc: # noqa: BLE001
failures.append((t.__name__, exc))
print(f"FAIL {t.__name__}: {exc}")
print(f"--- {len(tests) - len(failures)}/{len(tests)} passed")
return 0 if not failures else 1
if __name__ == "__main__":
sys.exit(_run())
pipelines/live_pipeline.py
+44 -2
View File
@@ -493,16 +493,20 @@ def process_one_message(
msg=msg,
cfg=papra_cfg,
)
papra_link: str | None = None
if papra_decision_obj.reason == "attachment_present":
papra_write_attachment(
written_ids = papra_write_attachment(
papra_cfg,
papra_decision_obj,
message_id=message_id,
module=None,
logger=logger,
)
papra_link = written_ids[0] if written_ids else None
if papra_link:
log_event(logger, "papra_uploaded", message_id=message_id, papra_link=papra_link)
elif papra_decision_obj.reason == "body_module":
papra_write_body(
papra_link = papra_write_body(
papra_cfg,
papra_decision_obj,
message_id=message_id,
@@ -522,6 +526,44 @@ def process_one_message(
dry_run=papra_cfg.dry_run,
)
# 4b. Papra_Link parity patch (REQ-INB-25). When the attachment
# walk returned a Papra document id, mirror the historical
# pipeline's behaviour and PATCH `Papra_Link` onto every
# module row that was upserted above. Idempotent: a re-run
# with the same papra_link produces no observable change.
if papra_link and upserts:
patched_modules: list[str] = []
for entry in upserts:
table_id = entry.get("table_id")
module_name = entry.get("module")
row_id = entry.get("id")
if not table_id or not row_id or not module_name:
continue
try:
nocodb_upsert(
nocodb_cfg,
table_id,
{"Papra_Link": papra_link},
message_id,
)
patched_modules.append(module_name)
except Exception as exc: # noqa: BLE001
log_event(
logger,
"papra_link_patch_failed",
message_id=message_id,
module=module_name,
error=str(exc),
)
if patched_modules:
log_event(
logger,
"papra_link_patched",
message_id=message_id,
papra_link=papra_link,
modules_patched=patched_modules,
)
log_event(logger, "process_done", message_id=message_id, upserts=upserts)
return True