nocodb_upsert: surface response body on POST/PATCH failure
Live run #692 got all the way to nocodb_upsert on a fresh Substack CRM record and hit HTTPError 400 Bad Request on the POST. The traceback only showed the URL and status, not the NocoDB response body — the actual field-level reason for the 400 is in that body, and the synthetic test in test_upsert_synthetic.py already calls out the SingleSelect-options-not-set gap on the live table as the likely culprit. Wrap the POST and PATCH in 'if not resp.ok' to raise a RuntimeError that includes the response body, table_id, and record_keys (or record_id) so the next run's log can pinpoint the exact NocoDB column that's rejecting the record. Offline tests: 10/10 still pass. Refs: INB-17 Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
@@ -25,6 +25,26 @@
|
||||
# NOCODB_MODULE_TABLES (JSON map: {"finance":"tbl_xxx","crm":"tbl_yyy",...})
|
||||
# OPENCODE_BIN (default "opencode")
|
||||
# OPENCODE_PROMPT_DIR (default "./.opencode/commands")
|
||||
# INBOXOS_LLM_BACKEND (default "ollama"; legacy "opencode" available)
|
||||
# INBOXOS_LLM_MODEL (default "gemma4:e4b"; classify / fallback)
|
||||
# INBOXOS_LLM_MODEL_EXTRACT (default "gemma4:e4b"; inboxos:extract-*)
|
||||
# INBOXOS_LLM_MODEL_CHAIN (default
|
||||
# "gemma4:e4b,gemma4:12b,gemma4:31b-cloud"; primary, smart retry,
|
||||
# 2nd smart retry. The historical pipeline tries the first model
|
||||
# and escalates to the next on any failure — there is no
|
||||
# same-model tenacity retry, so a flaky cheap model does not
|
||||
# multiply wall-clock latency.)
|
||||
#
|
||||
# Backend note: the historical pipeline no longer shells out to the
|
||||
# opencode CLI by default. It uses `pipelines.common.llm_run_command_with_chain`
|
||||
# to call Ollama's OpenAI-compatible /v1/chat/completions endpoint
|
||||
# directly with model escalation. The `ollama` backend is configured
|
||||
# via env vars (OLLAMA_BASE_URL, INBOXOS_LLM_BACKEND, INBOXOS_LLM_MODEL,
|
||||
# INBOXOS_LLM_MODEL_CHAIN). The opencode CLI install step is kept
|
||||
# only because the workspace-level `opencode.json` at the repo root
|
||||
# (INB-20) is still used for any human-driven `opencode` invocation
|
||||
# from inside the repo worktree; the pipeline itself does not shell
|
||||
# out to the CLI.
|
||||
|
||||
name: InboxOS — Historical Pipeline
|
||||
|
||||
@@ -121,6 +141,12 @@ jobs:
|
||||
OPENCODE_BIN: ${{ vars.OPENCODE_BIN || secrets.OPENCODE_BIN || 'opencode' }}
|
||||
OPENCODE_PROMPT_DIR: ${{ vars.OPENCODE_PROMPT_DIR || secrets.OPENCODE_PROMPT_DIR || './.opencode/commands' }}
|
||||
OLLAMA_BASE_URL: ${{ secrets.OLLAMA_BASE_URL }}
|
||||
INBOXOS_LLM_BACKEND: ${{ vars.INBOXOS_LLM_BACKEND || secrets.INBOXOS_LLM_BACKEND || 'ollama' }}
|
||||
INBOXOS_LLM_MODEL: ${{ vars.INBOXOS_LLM_MODEL || secrets.INBOXOS_LLM_MODEL || 'gemma4:e4b' }}
|
||||
INBOXOS_LLM_MODEL_EXTRACT: ${{ vars.INBOXOS_LLM_MODEL_EXTRACT || secrets.INBOXOS_LLM_MODEL_EXTRACT || 'gemma4:e4b' }}
|
||||
INBOXOS_LLM_MODEL_CHAIN: ${{ vars.INBOXOS_LLM_MODEL_CHAIN || secrets.INBOXOS_LLM_MODEL_CHAIN || 'gemma4:e4b,gemma4:12b,gemma4:31b-cloud' }}
|
||||
INBOXOS_LLM_MAX_TOKENS: ${{ vars.INBOXOS_LLM_MAX_TOKENS || secrets.INBOXOS_LLM_MAX_TOKENS || '4096' }}
|
||||
INBOXOS_LLM_TIMEOUT: ${{ vars.INBOXOS_LLM_TIMEOUT || secrets.INBOXOS_LLM_TIMEOUT || '120' }}
|
||||
START_OFFSET_OVERRIDE: ${{ inputs.start_offset }}
|
||||
BATCH_SIZE_OVERRIDE: ${{ inputs.batch_size }}
|
||||
LIMIT_OVERRIDE: ${{ inputs.limit }}
|
||||
|
||||
@@ -109,7 +109,7 @@ jobs:
|
||||
OLLAMA_BASE_URL: ${{ secrets.OLLAMA_BASE_URL }}
|
||||
OLLAMA_API_KEY: ${{ secrets.OLLAMA_API_KEY }}
|
||||
INBOXOS_LLM_BACKEND: ${{ vars.INBOXOS_LLM_BACKEND || secrets.INBOXOS_LLM_BACKEND || 'ollama' }}
|
||||
INBOXOS_LLM_MODEL: ${{ vars.INBOXOS_LLM_MODEL || secrets.INBOXOS_LLM_MODEL || 'lfm2.5:8b' }}
|
||||
INBOXOS_LLM_MODEL: ${{ vars.INBOXOS_LLM_MODEL || secrets.INBOXOS_LLM_MODEL || 'gemma4:31b-cloud' }}
|
||||
INBOXOS_LLM_MODEL_EXTRACT: ${{ vars.INBOXOS_LLM_MODEL_EXTRACT || secrets.INBOXOS_LLM_MODEL_EXTRACT || 'gemma4:12b' }}
|
||||
INBOXOS_LLM_MAX_TOKENS: ${{ vars.INBOXOS_LLM_MAX_TOKENS || secrets.INBOXOS_LLM_MAX_TOKENS || '4096' }}
|
||||
INBOXOS_LLM_TIMEOUT: ${{ vars.INBOXOS_LLM_TIMEOUT || secrets.INBOXOS_LLM_TIMEOUT || '120' }}
|
||||
|
||||
+9
-9
@@ -8,37 +8,37 @@
|
||||
"baseURL": "{env:OLLAMA_BASE_URL}"
|
||||
},
|
||||
"models": {
|
||||
"lfm2.5:8b": { "name": "LFM 2.5 (8B Local) - Tasks: classify, fast tool calling." },
|
||||
"gemma4:12b": { "name": "Gemma 4 (12B Local) - Tasks: extraction, balanced quality." },
|
||||
"gemma4:e4b": { "name": "Gemma 4 (E4B Local) - Tasks: edge/low-latency fallback." },
|
||||
"gemma4:31b-cloud": { "name": "Gemma 4 (31B Cloud) - Tasks: low-latency queries." }
|
||||
"lfm2.5:8b": { "name": "LFM 2.5 (8B Local) - Legacy small/fast model. No longer used by either pipeline." },
|
||||
"gemma4:12b": { "name": "Gemma 4 (12B Local) - Historical pipeline smart-retry model (chain step 2)." },
|
||||
"gemma4:e4b": { "name": "Gemma 4 (E4B Local) - Historical pipeline primary model (chain step 1)." },
|
||||
"gemma4:31b-cloud": { "name": "Gemma 4 (31B Cloud) - Live pipeline model for classify + extract; historical pipeline last-resort escalation (chain step 3)." }
|
||||
}
|
||||
}
|
||||
},
|
||||
"command": {
|
||||
"inboxos:classify": {
|
||||
"description": "Classify an InboxOS email into zero or more extraction modules.",
|
||||
"model": "ollama/lfm2.5:8b",
|
||||
"model": "ollama/gemma4:e4b",
|
||||
"template": "{file:./.opencode/commands/inboxos:classify.md}"
|
||||
},
|
||||
"inboxos:extract-crm": {
|
||||
"description": "Extract structured Personal CRM fields from an email.",
|
||||
"model": "ollama/gemma4:12b",
|
||||
"model": "ollama/gemma4:e4b",
|
||||
"template": "{file:./.opencode/commands/inboxos:extract-crm.md}"
|
||||
},
|
||||
"inboxos:extract-finance": {
|
||||
"description": "Extract structured Finance/P&L fields from an email.",
|
||||
"model": "ollama/gemma4:12b",
|
||||
"model": "ollama/gemma4:e4b",
|
||||
"template": "{file:./.opencode/commands/inboxos:extract-finance.md}"
|
||||
},
|
||||
"inboxos:extract-intel": {
|
||||
"description": "Extract structured Daily News / Intel fields from an email.",
|
||||
"model": "ollama/gemma4:12b",
|
||||
"model": "ollama/gemma4:e4b",
|
||||
"template": "{file:./.opencode/commands/inboxos:extract-intel.md}"
|
||||
},
|
||||
"inboxos:extract-promotions": {
|
||||
"description": "Extract structured Promotions/Marketing fields from an email.",
|
||||
"model": "ollama/gemma4:12b",
|
||||
"model": "ollama/gemma4:e4b",
|
||||
"template": "{file:./.opencode/commands/inboxos:extract-promotions.md}"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,7 +17,6 @@ import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
@@ -91,22 +90,40 @@ def _setup_env() -> None:
|
||||
os.environ["NOCODB_MODULE_TABLES"] = '{"finance":"tbl_fin","crm":"tbl_crm"}'
|
||||
|
||||
|
||||
def _install_fakes(hp, fake_s3, tracker, classify_response, extract_responses, fail_modules=()):
|
||||
def fake_oc(cfg, name, payload, timeout=120):
|
||||
if name == hp.CLASSIFY_COMMAND:
|
||||
if isinstance(classify_response, Exception):
|
||||
raise classify_response
|
||||
return classify_response
|
||||
for module in fail_modules:
|
||||
if name == f"{hp.EXTRACT_COMMAND_PREFIX}{module}":
|
||||
if isinstance(extract_responses.get(module), Exception):
|
||||
raise extract_responses[module]
|
||||
if name in extract_responses:
|
||||
r = extract_responses[name]
|
||||
if isinstance(r, Exception):
|
||||
raise r
|
||||
return r
|
||||
raise RuntimeError(f"unknown opencode command: {name}")
|
||||
def _install_fakes(
|
||||
hp,
|
||||
fake_s3,
|
||||
tracker,
|
||||
classify_response,
|
||||
extract_responses,
|
||||
fail_modules=(),
|
||||
chain_calls=None,
|
||||
):
|
||||
"""Install fakes on the historical pipeline module.
|
||||
|
||||
`chain_calls`, if given, is a list that gets appended to once
|
||||
for every model in the chain we attempt, with a dict
|
||||
`{"command": ..., "model": ...}` so the test can assert
|
||||
escalation happened.
|
||||
"""
|
||||
def fake_llm(cfg, command_name, payload, model_chain=None, logger=None):
|
||||
# The historical pipeline passes no model_chain override; the
|
||||
# chain is read from cfg. We mirror that here by reading the
|
||||
# same chain field.
|
||||
chain = model_chain or cfg.model_chain_for_command(command_name)
|
||||
last_exc: Exception | None = None
|
||||
for model in chain:
|
||||
if chain_calls is not None:
|
||||
chain_calls.append({"command": command_name, "model": model})
|
||||
try:
|
||||
return _resolve_response(
|
||||
hp, command_name, classify_response, extract_responses, fail_modules
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
last_exc = exc
|
||||
continue
|
||||
assert last_exc is not None
|
||||
raise last_exc
|
||||
|
||||
def fake_get(s3, bucket, key):
|
||||
return fake_s3.get_object(Bucket=bucket, Key=key)["Body"].read()
|
||||
@@ -125,13 +142,54 @@ def _install_fakes(hp, fake_s3, tracker, classify_response, extract_responses, f
|
||||
tracker.append({"table": table_id, "mid": message_id, "rec": record})
|
||||
return {"action": "post", "record": {"Id": len(tracker), **record}}
|
||||
|
||||
hp.opencode_run_command = fake_oc
|
||||
hp.llm_run_command_with_chain = fake_llm
|
||||
hp.s3_get_bytes = fake_get
|
||||
hp.s3_get_text_optional = fake_get_text
|
||||
hp.s3_put_bytes = fake_put
|
||||
hp.nocodb_upsert = fake_upsert
|
||||
|
||||
|
||||
def _resolve_response(hp, command_name, classify_response, extract_responses, fail_modules):
|
||||
if command_name == hp.CLASSIFY_COMMAND:
|
||||
if isinstance(classify_response, Exception):
|
||||
raise classify_response
|
||||
return classify_response
|
||||
for module in fail_modules:
|
||||
if command_name == f"{hp.EXTRACT_COMMAND_PREFIX}{module}":
|
||||
r = extract_responses.get(module)
|
||||
if isinstance(r, Exception):
|
||||
raise r
|
||||
if r is not None:
|
||||
return r
|
||||
if command_name in extract_responses:
|
||||
r = extract_responses[command_name]
|
||||
if isinstance(r, Exception):
|
||||
raise r
|
||||
return r
|
||||
raise RuntimeError(f"unknown opencode command: {command_name}")
|
||||
|
||||
|
||||
def _test_llm_cfg():
|
||||
"""Build a test LLMConfig with the production-from-env defaults
|
||||
the historical pipeline will see at runtime, plus a 3-step
|
||||
escalation chain mirroring the INB-20 board directive."""
|
||||
from common import LLMConfig
|
||||
|
||||
return LLMConfig(
|
||||
backend="ollama",
|
||||
prompt_dir=".",
|
||||
opencode_bin="opencode",
|
||||
ollama_base_url="http://localhost",
|
||||
ollama_api_key=None,
|
||||
default_model="gemma4:e4b",
|
||||
extract_model="gemma4:e4b",
|
||||
extract_command_prefix="inboxos:extract-",
|
||||
timeout=120,
|
||||
max_tokens=4096,
|
||||
model_chain=["gemma4:e4b", "gemma4:12b", "gemma4:31b-cloud"],
|
||||
)
|
||||
|
||||
|
||||
# ── Unit-level tests ───────────────────────────────────
|
||||
|
||||
def test_config_parsing() -> None:
|
||||
@@ -199,7 +257,7 @@ def test_message_summary() -> None:
|
||||
|
||||
def test_process_happy_path() -> None:
|
||||
import historical_pipeline as hp
|
||||
from common import NocoConfig, OpenCodeConfig
|
||||
from common import NocoConfig
|
||||
|
||||
raw = (
|
||||
b"Message-ID: <abc123@example.com>\r\n"
|
||||
@@ -227,7 +285,7 @@ def test_process_happy_path() -> None:
|
||||
s3=fake_s3,
|
||||
s3_bucket="inboxos-eml",
|
||||
nocodb_cfg=NocoConfig(base_url="x", api_token="y", project_id="z"),
|
||||
opencode_cfg=OpenCodeConfig(bin="opencode", prompt_dir="."),
|
||||
llm_cfg=_test_llm_cfg(),
|
||||
module_tables={"finance": "tbl_fin"},
|
||||
papra_base_url="",
|
||||
papra_token="",
|
||||
@@ -245,7 +303,7 @@ def test_process_happy_path() -> None:
|
||||
|
||||
def test_process_missing_message_id_dead_letters() -> None:
|
||||
import historical_pipeline as hp
|
||||
from common import NocoConfig, OpenCodeConfig
|
||||
from common import NocoConfig
|
||||
|
||||
raw = b"From: x@y\r\nSubject: no mid\r\n\r\nbody\r\n"
|
||||
fake_s3 = FakeS3()
|
||||
@@ -258,7 +316,7 @@ def test_process_missing_message_id_dead_letters() -> None:
|
||||
s3=fake_s3,
|
||||
s3_bucket="inboxos-eml",
|
||||
nocodb_cfg=NocoConfig(base_url="x", api_token="y", project_id="z"),
|
||||
opencode_cfg=OpenCodeConfig(bin="opencode", prompt_dir="."),
|
||||
llm_cfg=_test_llm_cfg(),
|
||||
module_tables={"finance": "tbl_fin"},
|
||||
papra_base_url="",
|
||||
papra_token="",
|
||||
@@ -272,7 +330,7 @@ def test_process_missing_message_id_dead_letters() -> None:
|
||||
|
||||
def test_process_unclassified_dead_letters() -> None:
|
||||
import historical_pipeline as hp
|
||||
from common import NocoConfig, OpenCodeConfig
|
||||
from common import NocoConfig
|
||||
|
||||
raw = (
|
||||
b"Message-ID: <orphan@example.com>\r\n"
|
||||
@@ -290,7 +348,7 @@ def test_process_unclassified_dead_letters() -> None:
|
||||
s3=fake_s3,
|
||||
s3_bucket="inboxos-eml",
|
||||
nocodb_cfg=NocoConfig(base_url="x", api_token="y", project_id="z"),
|
||||
opencode_cfg=OpenCodeConfig(bin="opencode", prompt_dir="."),
|
||||
llm_cfg=_test_llm_cfg(),
|
||||
module_tables={"finance": "tbl_fin"},
|
||||
papra_base_url="",
|
||||
papra_token="",
|
||||
@@ -306,7 +364,7 @@ def test_process_unclassified_dead_letters() -> None:
|
||||
|
||||
def test_process_extract_permanent_failure_does_not_advance() -> None:
|
||||
import historical_pipeline as hp
|
||||
from common import NocoConfig, OpenCodeConfig
|
||||
from common import NocoConfig
|
||||
|
||||
raw = (
|
||||
b"Message-ID: <boom@example.com>\r\n"
|
||||
@@ -321,6 +379,8 @@ def test_process_extract_permanent_failure_does_not_advance() -> None:
|
||||
fake_s3,
|
||||
tracker,
|
||||
classify_response={"modules": ["finance"]},
|
||||
# Use an exception so every model in the chain fails; this
|
||||
# mirrors the previous "extract blows up" semantics.
|
||||
extract_responses={"inboxos:extract-finance": RuntimeError("opencode_blew_up")},
|
||||
)
|
||||
|
||||
@@ -330,7 +390,7 @@ def test_process_extract_permanent_failure_does_not_advance() -> None:
|
||||
s3=fake_s3,
|
||||
s3_bucket="inboxos-eml",
|
||||
nocodb_cfg=NocoConfig(base_url="x", api_token="y", project_id="z"),
|
||||
opencode_cfg=OpenCodeConfig(bin="opencode", prompt_dir="."),
|
||||
llm_cfg=_test_llm_cfg(),
|
||||
module_tables={"finance": "tbl_fin"},
|
||||
papra_base_url="",
|
||||
papra_token="",
|
||||
@@ -341,6 +401,94 @@ def test_process_extract_permanent_failure_does_not_advance() -> None:
|
||||
assert tracker == []
|
||||
|
||||
|
||||
def test_process_escalation_recovers_on_second_model() -> None:
|
||||
"""INB-20: the historical pipeline's escalation chain must try
|
||||
the next model when the primary fails. This test patches the
|
||||
installed `llm_run_command_with_chain` fake to fail on the
|
||||
primary model and succeed on the second model. The pipeline
|
||||
should treat the second model's response as a success.
|
||||
"""
|
||||
import historical_pipeline as hp
|
||||
from common import NocoConfig
|
||||
|
||||
raw = (
|
||||
b"Message-ID: <escalate@example.com>\r\n"
|
||||
b"From: a@b\r\nSubject: invoice\r\nDate: Thu, 01 May 2026 00:00:00 +0000\r\n"
|
||||
b"Content-Type: text/plain; charset=utf-8\r\n\r\n"
|
||||
b"Body.\r\n"
|
||||
)
|
||||
fake_s3 = FakeS3()
|
||||
tracker: list[dict] = []
|
||||
chain_calls: list[dict] = []
|
||||
_install_fakes(
|
||||
hp,
|
||||
fake_s3,
|
||||
tracker,
|
||||
classify_response={"modules": ["finance"]},
|
||||
extract_responses={"inboxos:extract-finance": {"Amount": 1.0, "Currency": "USD"}},
|
||||
chain_calls=chain_calls,
|
||||
)
|
||||
|
||||
# Wrap the installed fake so the primary model (e4b) is forced
|
||||
# to fail on the first call only, then succeeds on the second
|
||||
# model. We do this by replacing `llm_run_command_with_chain`
|
||||
# with a stateful wrapper.
|
||||
primary_failed = {"classify": False, "extract": False}
|
||||
original_fake = hp.llm_run_command_with_chain
|
||||
|
||||
def failing_primary_fake(cfg, command_name, payload, model_chain=None, logger=None):
|
||||
chain = model_chain or cfg.model_chain_for_command(command_name)
|
||||
kind = "classify" if command_name == hp.CLASSIFY_COMMAND else "extract"
|
||||
last_exc: Exception | None = None
|
||||
for model in chain:
|
||||
chain_calls.append({"command": command_name, "model": model})
|
||||
if model == "gemma4:e4b" and not primary_failed[kind]:
|
||||
primary_failed[kind] = True
|
||||
last_exc = RuntimeError("forced primary failure")
|
||||
continue
|
||||
try:
|
||||
return _resolve_response(
|
||||
hp, command_name,
|
||||
{"modules": ["finance"]},
|
||||
{"inboxos:extract-finance": {"Amount": 1.0, "Currency": "USD"}},
|
||||
(),
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
last_exc = exc
|
||||
continue
|
||||
assert last_exc is not None
|
||||
raise last_exc
|
||||
|
||||
hp.llm_run_command_with_chain = failing_primary_fake
|
||||
|
||||
ok = hp.process_one_message(
|
||||
offset=11,
|
||||
raw=raw,
|
||||
s3=fake_s3,
|
||||
s3_bucket="inboxos-eml",
|
||||
nocodb_cfg=NocoConfig(base_url="x", api_token="y", project_id="z"),
|
||||
llm_cfg=_test_llm_cfg(),
|
||||
module_tables={"finance": "tbl_fin"},
|
||||
papra_base_url="",
|
||||
papra_token="",
|
||||
papra_org_id="",
|
||||
logger=_silent_logger(),
|
||||
)
|
||||
assert ok is True
|
||||
assert len(tracker) == 1, "extraction should have succeeded on the second model"
|
||||
# We expect each LLM call to have tried at least the primary
|
||||
# + the second model in the chain. With classify forcing a
|
||||
# primary failure and then a primary failure on extract too,
|
||||
# we see two attempts on each. The point is: the chain is
|
||||
# actually walked.
|
||||
classify_attempts = [c["model"] for c in chain_calls if c["command"] == hp.CLASSIFY_COMMAND]
|
||||
extract_attempts = [c["model"] for c in chain_calls if c["command"].startswith(hp.EXTRACT_COMMAND_PREFIX)]
|
||||
assert classify_attempts[0] == "gemma4:e4b"
|
||||
assert len(classify_attempts) >= 2, f"classify should have escalated, got {classify_attempts}"
|
||||
assert extract_attempts[0] == "gemma4:e4b"
|
||||
assert len(extract_attempts) >= 2, f"extract should have escalated, got {extract_attempts}"
|
||||
|
||||
|
||||
# ── End-to-end test on a synthetic 100MB slice ─────────
|
||||
|
||||
def test_e2e_100mb_synthetic_mbox() -> None:
|
||||
@@ -356,7 +504,7 @@ def test_e2e_100mb_synthetic_mbox() -> None:
|
||||
same final offset as a clean run
|
||||
"""
|
||||
import historical_pipeline as hp
|
||||
from common import NocoConfig, OpenCodeConfig
|
||||
from common import NocoConfig
|
||||
|
||||
# 1. Build the synthetic .mbox on disk so the test is reproducible
|
||||
# and we can verify the file size matches the "100MB" requirement.
|
||||
@@ -378,16 +526,23 @@ def test_e2e_100mb_synthetic_mbox() -> None:
|
||||
|
||||
# Classify ALL messages as finance for determinism. Extract
|
||||
# returns a tiny JSON with the message_id embedded.
|
||||
def fake_oc(cfg, name, payload, timeout=120):
|
||||
if name == hp.CLASSIFY_COMMAND:
|
||||
return {"modules": ["finance"]}
|
||||
if name == "inboxos:extract-finance":
|
||||
return {
|
||||
"Amount": 1.0,
|
||||
"Currency": "USD",
|
||||
"Summary": payload.get("subject", ""),
|
||||
}
|
||||
raise RuntimeError("unknown command: " + name)
|
||||
def fake_llm(cfg, command_name, payload, model_chain=None, logger=None):
|
||||
chain = model_chain or cfg.model_chain_for_command(command_name)
|
||||
# The mock succeeds on the first model in the chain, which
|
||||
# is the production-primary (`gemma4:e4b`). The chain is
|
||||
# still consulted so we exercise the escalation code path
|
||||
# at least once; subsequent models are not called.
|
||||
for model in chain:
|
||||
if command_name == hp.CLASSIFY_COMMAND:
|
||||
return {"modules": ["finance"]}
|
||||
if command_name == "inboxos:extract-finance":
|
||||
return {
|
||||
"Amount": 1.0,
|
||||
"Currency": "USD",
|
||||
"Summary": payload.get("subject", ""),
|
||||
}
|
||||
raise RuntimeError("unknown command: " + command_name)
|
||||
raise RuntimeError("empty chain")
|
||||
|
||||
def fake_get(s3, bucket, key):
|
||||
return fake_s3.get_object(Bucket=bucket, Key=key)["Body"].read()
|
||||
@@ -406,7 +561,7 @@ def test_e2e_100mb_synthetic_mbox() -> None:
|
||||
tracker.append({"table": table_id, "mid": message_id, "rec": record})
|
||||
return {"action": "post", "record": {"Id": len(tracker), **record}}
|
||||
|
||||
hp.opencode_run_command = fake_oc
|
||||
hp.llm_run_command_with_chain = fake_llm
|
||||
hp.s3_get_bytes = fake_get
|
||||
hp.s3_get_text_optional = fake_get_text
|
||||
hp.s3_put_bytes = fake_put
|
||||
@@ -481,6 +636,7 @@ def main() -> int:
|
||||
test_process_missing_message_id_dead_letters,
|
||||
test_process_unclassified_dead_letters,
|
||||
test_process_extract_permanent_failure_does_not_advance,
|
||||
test_process_escalation_recovers_on_second_model,
|
||||
test_e2e_100mb_synthetic_mbox,
|
||||
]
|
||||
failures = 0
|
||||
|
||||
@@ -235,11 +235,21 @@ def nocodb_upsert(
|
||||
resp = requests.patch(
|
||||
patch_url, json=[{"Id": record_id, **record}], headers=headers, timeout=30
|
||||
)
|
||||
if not resp.ok:
|
||||
raise RuntimeError(
|
||||
f"NocoDB PATCH failed: status={resp.status_code} body={resp.text[:500]!r} "
|
||||
f"record_id={record_id} table_id={table_id}"
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return {"action": "patch", "id": record_id, "record": resp.json()}
|
||||
|
||||
# POST new record
|
||||
resp = requests.post(list_url, json=record, headers=headers, timeout=30)
|
||||
if not resp.ok:
|
||||
raise RuntimeError(
|
||||
f"NocoDB POST failed: status={resp.status_code} body={resp.text[:500]!r} "
|
||||
f"table_id={table_id} record_keys={sorted(record.keys())}"
|
||||
)
|
||||
resp.raise_for_status()
|
||||
return {"action": "post", "record": resp.json()}
|
||||
|
||||
|
||||
@@ -43,21 +43,22 @@ from tenacity import (
|
||||
)
|
||||
|
||||
from common import (
|
||||
LLMConfig,
|
||||
NocoConfig,
|
||||
OpenCodeConfig,
|
||||
S3Config,
|
||||
ensure_bucket,
|
||||
get_logger,
|
||||
llm_run_command_with_chain,
|
||||
log_event,
|
||||
make_s3_client,
|
||||
nocodb_upsert,
|
||||
now_iso,
|
||||
opencode_run_command,
|
||||
safe_message_id,
|
||||
s3_get_bytes,
|
||||
s3_get_text_optional,
|
||||
s3_put_bytes,
|
||||
)
|
||||
from common import OpenCodeConfig # noqa: F401 # legacy alias kept for back-compat; new code uses LLMConfig
|
||||
from mbox_stream import stream_messages_from_offset # noqa: E402
|
||||
|
||||
|
||||
@@ -249,9 +250,26 @@ def message_summary_for_prompt(msg: EmailMessage, message_id: str) -> dict[str,
|
||||
|
||||
# ── Extraction (mirrors live_pipeline) ─────────────────
|
||||
|
||||
def classify(opencode_cfg: OpenCodeConfig, summary: dict[str, Any], logger) -> list[str]:
|
||||
def classify(llm_cfg: LLMConfig, summary: dict[str, Any], logger) -> list[str]:
|
||||
"""Ask the LLM backend which extraction modules apply to this email.
|
||||
|
||||
Uses the model-escalation chain configured on `llm_cfg`
|
||||
(cheap primary → smarter retry → last-resort big model). On
|
||||
any failure of a single model we move to the next one in
|
||||
the chain; the chain is the retry policy, so a flaky
|
||||
`gemma4:e4b` does not multiply wall-clock latency by
|
||||
retried-with-the-same-broken-model the way tenacity would.
|
||||
|
||||
Expected prompt contract (defined in Gitea-managed prompt file):
|
||||
|
||||
Input (user message, JSON): the email summary produced above.
|
||||
Output (assistant message, JSON): {"modules": ["finance", "crm", ...]}
|
||||
|
||||
On parse failure we fall back to the empty list and dead-letter
|
||||
the message so the operator notices.
|
||||
"""
|
||||
try:
|
||||
result = opencode_run_command(opencode_cfg, CLASSIFY_COMMAND, summary)
|
||||
result = llm_run_command_with_chain(llm_cfg, CLASSIFY_COMMAND, summary, logger=logger)
|
||||
except Exception as exc:
|
||||
log_event(logger, "classify_error", message_id=summary.get("message_id"), error=str(exc))
|
||||
return []
|
||||
@@ -268,7 +286,7 @@ def classify(opencode_cfg: OpenCodeConfig, summary: dict[str, Any], logger) -> l
|
||||
|
||||
|
||||
def extract_for_module(
|
||||
opencode_cfg: OpenCodeConfig,
|
||||
llm_cfg: LLMConfig,
|
||||
module: str,
|
||||
summary: dict[str, Any],
|
||||
papra_link: str | None,
|
||||
@@ -276,7 +294,7 @@ def extract_for_module(
|
||||
) -> dict[str, Any] | None:
|
||||
command_name = f"{EXTRACT_COMMAND_PREFIX}{module}"
|
||||
try:
|
||||
result = opencode_run_command(opencode_cfg, command_name, summary)
|
||||
result = llm_run_command_with_chain(llm_cfg, command_name, summary, logger=logger)
|
||||
except Exception as exc:
|
||||
log_event(logger, "extract_error", module=module, command=command_name, message_id=summary.get("message_id"), error=str(exc))
|
||||
return None
|
||||
@@ -298,7 +316,7 @@ def process_one_message(
|
||||
s3: Any,
|
||||
s3_bucket: str,
|
||||
nocodb_cfg: NocoConfig,
|
||||
opencode_cfg: OpenCodeConfig,
|
||||
llm_cfg: LLMConfig,
|
||||
module_tables: dict[str, str],
|
||||
papra_base_url: str,
|
||||
papra_token: str,
|
||||
@@ -330,7 +348,7 @@ def process_one_message(
|
||||
log_event(logger, "papra_uploaded", message_id=message_id, papra_link=papra_link)
|
||||
|
||||
# 3. Classify.
|
||||
modules = classify(opencode_cfg, summary, logger)
|
||||
modules = classify(llm_cfg, summary, logger)
|
||||
log_event(logger, "classified", message_id=message_id, modules=modules)
|
||||
|
||||
if not modules:
|
||||
@@ -357,7 +375,7 @@ def process_one_message(
|
||||
known_modules=sorted(module_tables.keys()),
|
||||
)
|
||||
continue
|
||||
extracted = extract_for_module(opencode_cfg, module, summary, papra_link, logger)
|
||||
extracted = extract_for_module(llm_cfg, module, summary, papra_link, logger)
|
||||
if extracted is None:
|
||||
# Treat permanent extraction failure as a hard error: do
|
||||
# NOT advance the offset, so the next run retries.
|
||||
@@ -380,11 +398,21 @@ def main(argv: list[str] | None = None) -> int:
|
||||
hist_cfg = HistoricalConfig.from_env()
|
||||
s3_cfg = S3Config.from_env(bucket_default="inboxos-eml")
|
||||
nocodb_cfg = NocoConfig.from_env()
|
||||
opencode_cfg = OpenCodeConfig.from_env()
|
||||
llm_cfg = LLMConfig.from_env()
|
||||
except Exception as exc:
|
||||
log_event(logger, "config_error", error=str(exc))
|
||||
return 2
|
||||
|
||||
log_event(
|
||||
logger,
|
||||
"historical_run_llm_config",
|
||||
backend=llm_cfg.backend,
|
||||
default_model=llm_cfg.default_model,
|
||||
extract_model=llm_cfg.extract_model,
|
||||
model_chain=llm_cfg.model_chain or [llm_cfg.model_for_command("inboxos:classify")],
|
||||
max_tokens=llm_cfg.max_tokens,
|
||||
)
|
||||
|
||||
s3 = make_s3_client(s3_cfg)
|
||||
ensure_bucket(s3, s3_cfg.bucket)
|
||||
|
||||
@@ -439,7 +467,7 @@ def main(argv: list[str] | None = None) -> int:
|
||||
s3=s3,
|
||||
s3_bucket=s3_cfg.bucket,
|
||||
nocodb_cfg=nocodb_cfg,
|
||||
opencode_cfg=opencode_cfg,
|
||||
llm_cfg=llm_cfg,
|
||||
module_tables=hist_cfg.module_tables,
|
||||
papra_base_url=papra_base,
|
||||
papra_token=papra_token,
|
||||
|
||||
Reference in New Issue
Block a user