diff --git a/.gitea/workflows/historical-pipeline.yml b/.gitea/workflows/historical-pipeline.yml index e3f8884..e3c2ccf 100644 --- a/.gitea/workflows/historical-pipeline.yml +++ b/.gitea/workflows/historical-pipeline.yml @@ -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 }} diff --git a/.gitea/workflows/live-pipeline.yml b/.gitea/workflows/live-pipeline.yml index 20a3248..8f64ec4 100644 --- a/.gitea/workflows/live-pipeline.yml +++ b/.gitea/workflows/live-pipeline.yml @@ -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' }} diff --git a/opencode.json b/opencode.json index 4c480d1..f72eb26 100644 --- a/opencode.json +++ b/opencode.json @@ -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}" } } diff --git a/pipelines/_tests/test_historical_pipeline_offline.py b/pipelines/_tests/test_historical_pipeline_offline.py index c6edd0d..d2af8d5 100644 --- a/pipelines/_tests/test_historical_pipeline_offline.py +++ b/pipelines/_tests/test_historical_pipeline_offline.py @@ -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: \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: \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: \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: \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 diff --git a/pipelines/common.py b/pipelines/common.py index 005de26..3a8e8d9 100644 --- a/pipelines/common.py +++ b/pipelines/common.py @@ -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()} diff --git a/pipelines/historical_pipeline.py b/pipelines/historical_pipeline.py index 70360f1..f5e9a2c 100644 --- a/pipelines/historical_pipeline.py +++ b/pipelines/historical_pipeline.py @@ -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,