inboxos-v1/pipelines

InboxOS — Pipelines

This directory contains the version-controlled Python scripts that the Gitea Actions workflows invoke. Workflows live in .gitea/workflows/ and are the only orchestration layer; this is the explicit board decision recorded on INB-8 (n8n was ruled out as unreliable).

Files

File	Used by	Purpose
common.py	both	Shared helpers: S3, NocoDB UPSERT, OpenCode invocation, structured logging.
mbox_stream.py	historical	Byte-offset streaming parser for .mbox files.
live_pipeline.py	live-pipeline.yml	Polls IMAP, archives .eml, calls OpenCode, UPSERTs to NocoDB.
historical_pipeline.py	historical-pipeline.yml	Streams .mbox, uploads attachments to Papra, calls OpenCode, UPSERTs to NocoDB.

Idempotency model

Both pipelines share the same idempotency contract. Retries and re-runs are safe by construction.

• NocoDB writes go through common.nocodb_upsert which looks up the record by Message_ID and PATCHes if present, POSTs otherwise. There is no INSERT path in the codebase.
• IMAP high-water mark (live) is persisted to s3://$SEAWEEDFS_S3_BUCKET/state/live-state.json after every run. A failed run does not advance the mark past a failed UID, so the next run retries it.
• .mbox byte offset (historical) is persisted to s3://$SEAWEEDFS_S3_BUCKET/state/historical-state.json after every checkpoint batch. A run can resume mid-file.
• Object keys are deterministic by Message-ID and date, so re-uploading the same .eml overwrites the same S3 object.

Why Gitea Actions (not n8n)

The board explicitly rejected n8n on the grounds that workflow orchestration should be version-controlled in the same repository as the scripts and prompts it runs, with no separate runtime state to back up. Gitea Actions gives us:

• Workflows as .yml in git, with diffs and PRs.
• Self-hosted runner (gitea-runner service) — no external SaaS.
• Per-job secrets, concurrency groups, schedules, and manual dispatch all in one declarative file.
• Standard CI/CD mental model that any engineer can read.

Operational notes

• pipelines/_artifacts/ is the per-run log directory. The workflows upload any logs there as build artifacts.
• Both pipelines emit structured JSON to stdout (one event per line). Gitea Actions surfaces this in the workflow run UI.
• All retries use tenacity with exponential backoff; transient S3 / NocoDB / OpenCode failures are handled in common.py.
• Two self-hosted runners are registered (inboxos-runner-1 and inboxos-runner-2, both labelled self-hosted,linux,x64,gitea-runner:docker). Live and historical pipelines can therefore execute in parallel: the every-5-min live cron does not block a workflow_dispatch run of the historical pipeline (and vice versa). Provisioned by INB-33; see docs/gitea-actions.md §2a for registration steps.

Local development

The scripts can be run by hand against a local stack:

export $(grep -v '^#' .env | xargs)
python pipelines/live_pipeline.py

For the historical pipeline, set MBOX_OBJECT_KEY to the key of an .mbox file already uploaded to SeaweedFS.

For the live pipeline there is an offline smoke that exercises parsing, key generation, dead-letter, and state round-trip with no external services:

python pipelines/_tests/test_live_pipeline_offline.py

Live pipeline — flow

┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│ Load checkpoint  │ ──▶ │ Poll IMAP for    │ ──▶ │ For each message │
│ state/live-      │     │ UID > last_uid   │     │ (UID ascending)  │
│  state.json      │     │ (bounded by      │     │                  │
└──────────────────┘     │  MAX_EMAILS=50)  │     └────────┬─────────┘
                         └──────────────────┘              │
                                                           ▼
┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│ UPSERT into      │ ◀── │ Per-module       │ ◀── │ Classify via     │
│ NocoDB (xc-token)│     │ extract via      │     │ OpenCode/Ollama  │
│ on Message-ID    │     │ OpenCode/Ollama  │     │                  │
└────────┬─────────┘     └──────────────────┘     └──────────────────┘
         │
         ▼
┌──────────────────┐
│ Save checkpoint  │  (per-message; a crash never replays a success)
└──────────────────┘

Failure handling (live)

Step	Failure mode	Behavior
IMAP fetch	Transient	Tenacity retries 3x exponential. Permanent failure exits non-zero.
Missing Message-ID	No idempotency key available	Raw .eml dead-lettered to eml/dead-letter/uid-<n>.eml; cursor advances.
SeaweedFS write	Transient	Tenacity retries 5x.
OpenCode classify	Parse failure / modules: []	Dead-lettered as eml/dead-letter/<mid>.eml (JSON marker); cursor advances.
OpenCode extract	Permanent failure	Pipeline stops at this UID; next run retries; UPSERT idempotency keeps NocoDB clean.
NocoDB write	Transient	Tenacity retries 5x. Permanent failure stops the run.
Unknown module	Module not in NOCODB_MODULE_TABLES	Logged; other modules still extracted; re-run after schemas land catches it.

Required Gitea repo secrets (Settings → Actions → Secrets)

# IMAP — Gmail (set up an app-password if 2FA is on)
IMAP_HOST=imap.gmail.com
IMAP_PORT=993
IMAP_USER=<gmail-address>
IMAP_PASSWORD=<app-password>
IMAP_USE_SSL=true
IMAP_MAILBOX=INBOX

# SeaweedFS (S3-compatible, in-network)
SEAWEEDFS_S3_ENDPOINT=http://seaweedfs-filer:8888
SEAWEEDFS_S3_ACCESS_KEY=<seaweed-access-key>
SEAWEEDFS_S3_SECRET_KEY=<seaweed-secret-key>
SEAWEEDFS_S3_BUCKET=inboxos

# NocoDB  (canonical URL — see INB-9 verification)
NOCODB_BASE_URL=https://noco.cloudinit.dev
NOCODB_API_TOKEN=<xc-token>
NOCODB_PROJECT_ID=<nocodb-project-id>
NOCODB_MODULE_TABLES={"finance":"tbl_xxx","crm":"tbl_yyy","intel":"tbl_zzz","promotions":"tbl_www"}

# OpenCode + Ollama  (canonical URL — see INB-9 verification)
OPENCODE_BIN=opencode
OPENCODE_PROMPT_DIR=./.opencode/commands
OLLAMA_BASE_URL=https://ollama-frp.cloudinit.dev/v1
# Canonical form: the URL INCLUDES the /v1 suffix; pipelines/common.py
# ::LLMConfig.from_env appends /chat/completions to whatever is supplied
# (see LLMConfig.from_env at pipelines/common.py:463-465).

NOCODB_MODULE_TABLES is populated by INB-4 once the four module tables exist. The pipeline tolerates an empty map at run time — unknown modules are logged and the .eml is preserved in SeaweedFS for a re-run after the schemas land.

Rollback

If a prompt change produces bad data:

1. Revert the prompt commit in Gitea. The next 5-minute cron uses the reverted prompt automatically because workflows run from the head commit.
2. Re-run affected messages by dispatching live-pipeline.yml manually with since_uid set to the lowest affected UID. UPSERT on Message-ID overwrites the bad NocoDB rows in place.
3. For data poisoning concerns, the raw .eml in s3://$SEAWEEDFS_S3_BUCKET/eml/live/<yyyy>/<mm>/<dd>/<mid>.eml is the source of truth — re-extracting from that file regenerates the NocoDB row deterministically.

To pause the pipeline entirely, disable the workflow in Gitea → repo → Actions, or push a commit that comments out the schedule: trigger.

OpenCode prompts

The five prompt files live under .opencode/commands/:

• inboxos:classify.md — routes each email to zero or more modules.
• inboxos:extract-finance.md
• inboxos:extract-crm.md
• inboxos:extract-intel.md
• inboxos:extract-promotions.md

All five reference model ollama/gemma4:e4b — confirmed available on the live Ollama (https://ollama-frp.cloudinit.dev/api/tags). This is the canonical pin per REQ-INB-11; the model is single-pinned with no escalation chain. Prompts are version-controlled so edits go through git review and never silently corrupt extracted data — re-running the pipeline deterministically refreshes any NocoDB record.

Historical pipeline — flow

┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│ Load offset      │ ──▶ │ Download .mbox   │ ──▶ │ stream_messages  │
│ state/historical │     │ from SeaweedFS   │     │ _from_offset     │
│  -state.json     │     │ MBOX_OBJECT_KEY  │     │ (mbox_stream.py) │
└──────────────────┘     └──────────────────┘     └────────┬─────────┘
                                                           │
                                                           ▼
┌──────────────────┐     ┌──────────────────┐     ┌──────────────────┐
│ Save checkpoint  │ ◀── │ UPSERT each      │ ◀── │ For each message:│
│ every BATCH_SIZE │     │ module to NocoDB │     │ backup .eml →    │
│ messages         │     │ on Message-ID    │     │ upload attach-   │
└──────────────────┘     └──────────────────┘     │ ments to Papra → │
                                                   │ classify → extract│
                                                   └──────────────────┘

Failure handling (historical)

Same model as the live pipeline. A permanent extraction failure on one message stops the run without advancing the offset, so the next run retries the same message. A streaming failure flushes whatever checkpoint we already have and exits non-zero.

Required Gitea repo secrets (additional, beyond live)

MBOX_BUCKET=inboxos                       # defaults to SEAWEEDFS_S3_BUCKET
MBOX_OBJECT_KEY=eml/historical/all-mail.mbox # passed as workflow_dispatch input
PAPRA_BASE_URL=https://papra.cloudinit.dev
PAPRA_API_TOKEN=<papra-api-token>
PAPRA_ORGANIZATION_ID=<papra-org-id>

MBOX_OBJECT_KEY is the workflow_dispatch input; the other three are repository secrets. Papra secrets can be left blank — the historical pipeline logs papra_attachment_failed and continues without Papra document links in that case.

Rollback (historical)

• Revert a bad prompt, then re-run the workflow. UPSERT overwrites the same Message-ID row in place.
• For an archive that has bad data, re-run with START_OFFSET_OVERRIDE=<offset-of-bad-batch> and UPSERT deterministically refreshes the rows.
• To restart from scratch, delete state/historical-state.json from the SeaweedFS bucket.