Signals harness: full design

1. Goal and scope ↑ top

Every "read a deal's activity, ask the model a question, write the answer back" job runs on one harness. The harness does the hard, shared parts once: scheduling, locking, knowing when data is complete, skipping unchanged items, calling the model, checking quotes, writing safely, and tracking cost. Each job is a small plugin that only knows its question and its rules.

2. Architecture flow ↑ top

3. Sources: what the harness reads ↑ top

The harness never calls a provider. Oliv's ingestion already lands every source in Baserow with Deal/Company links and its own retries. The harness's job is narrower: notice what is new or changed per deal, and wait until a day's data is complete.

Adapter contract

One adapter per source. Each reads one Oliv table and returns normalized items.

Item {
  ref           "email:123" | "meeting:456" | "slack:789" | "ticket:42"   stable Oliv row
  source_id     provider identity (Message id, channel+ts, ClickUp id)
  kind          email | meeting | slack | ticket
  deal_ids, company_ids
  occurred_at   when it happened
  changed_at    Oliv-side change stamp
  voice         customer | internal | mixed
  text          what the model reads
  content_hash  sha256(text + links + status): a change means re-evaluate
  ready         False = known but not evaluable yet (e.g. no transcript)
}

SourceAdapter:
  changed(ws, since, until)   cheap: candidates by change stamp
  for_subject(ws, subject)    complete: every item linked to a deal/account
  completeness(ws, day)       complete | partial | failed

How changes are found (no schema change)

4. Source Config ↑ top

One Baserow row per org × source. Says which Oliv tables the org has and how to read them. No credentials field.

5. Completeness: when a day counts ↑ top

Before any signal reads a day, the harness records each source's status for that org and day: complete, partial or failed.

• Rule: every enabled source must be complete before deciding. A signal can relax this with Sources in its Signal Config row.
• Partial: no decision for that subject, old output stays, watermark doesn't move, retried next run. After 3 days the run report shows "evidence incomplete".
• Weekly count check (built in, off by default): a separate ingestion-side job compares Slack and ClickUp counts for the past 7 days with what's in Oliv. Over 1% gap → source marked suspect (counts as not complete) and ingestion's reconcile is queued.

6. Signals (plugins) and their interface ↑ top

A plugin is one small Python module. Everything with I/O stays in the core.

class Signal(Protocol):
    name: str                          # "competitors"
    subject: "deal" | "account"
    sources: set[str]                  # {"meeting", "email"}
    settings_schema: dict              # shape of the org's Settings cell
    output_schema: dict                # what the model must return

    def prompt(subject, items, current, settings) -> str
    def decide(readings, items, current, settings) -> Decision   # pure, no I/O

Decision(
    rows=[...],        # child rows, e.g. one per (deal, competitor)
    values={...},      # named outputs, e.g. {"health": "Escalation", "description": ...}
)

The two plugins in this PR

A new signal (escalation reasons, next steps, …) = a new plugin folder + a config row. The rule: only extract a shared piece once two plugins use it.

7. Where prompts live ↑ top

Each prompt has two parts: fixed rules in code (same for every org, versioned and tested) and org-specific text in Baserow (data the prompt reads).

8. Signal Config ↑ top

One Baserow row per org × signal. The plugin is the question; the row says for which org, how hard to push, and where the answers go. Replaces config/deal_competitors.json and the old per-org flags.

9. Modes: off, dry-run, live ↑ top

10. How one run works, step by step ↑ top

1. Dispatch reads every Signal Config row whose mode isn't off.
2. Validate Settings and Outputs against the plugin. Invalid → that org × signal is skipped, Runs row says failed: config invalid, nothing written.
3. Snapshot the row. An edit mid-run applies next run; one run never mixes two configs.
4. Pick subjects: deals/accounts with candidate items since the watermark (with overlap). Weekly: every open subject.
5. Lease per (org, signal, subject). A second delivery sees running/done and exits.
6. Read items through the adapters.
7. Completeness gate: any required source partial for the window → no decision, old output stays.
8. Ledger diff: only items whose hash is new go on. Nothing changed → no model call. Not ready → kept as pending, no time limit.
9. Model: plugin.prompt() → Codex turn with a shared cached prefix per org → facts with {ref, quote}.
10. Quote check: drop any fact whose quote (numbers and dates too) isn't in the cited item.
11. Decide: plugin.decide(), pure rules → values and rows.
12. Publish each output with its policy. Lease checked before every Baserow send, retries included. Person edited since read → held. Readback after write. Dry-run: record only.
13. Ledger written last, then the watermark moves. Runs row updated with counts, cost, errors. Limits checked between subjects.

11. Write policies ↑ top

12. Durability and crash safety ↑ top

13. Where state lives ↑ top

14. Code layout ↑ top

services/signals/
  dispatch.py       beat → org × signal → subjects
  config.py         read + validate Source/Signal Config rows, snapshot
  lease.py          fenced lease, kill switch
  sources/          email, meeting, slack, ticket adapters + completeness
  ledger.py         S3 item hashes, pending, watermark
  runner.py         Codex turn, cached prefix, cost per run
  verify.py         quote check
  publish/          card, rows, if-empty, overwrite, review, readback
  runs.py           Runs rows (Type = signal)
  plugins/
    field_extraction/   prompt.py, schema, decide()
    competitors/        prompt.py, schema, decide()
jobs/tasks/signals.py   Celery entry points (harness pool)

15. Onboarding and backfill a new org ↑ top

1. Ingestion backfills first with existing tools (oliv-initial-backfill for meetings/emails, Slack full-history pull, ClickUp/Jira importers).
2. Add Source Config rows; wait until completeness reads complete for the whole range.
3. Add a Signal Config row in dry-run with a lookback (e.g. 180 days) and a $ cap.
4. Harness runs with an empty ledger, oldest first. Review the dry-run Runs rows.
5. Flip Mode to live.

16. Test plan ↑ top

Workflow-sim (real adapter, ledger, publisher code on a virtual clock; Baserow modelled from real column shapes), kept local. Each case runs against today's code as baseline and the new path, asserting final card/rows/fields and ledger contents.

17. Done: when the old PRs close ↑ top

1. Competitors parity: dry run on the same Onyx deals; card values and rows match the old competitors code exactly (except the two fixed bugs, cases 1 and 2).
2. Field extraction parity: replay the Qu and Whatfix shadow sets; same values, same or fewer wrong, cost no higher.
3. Safety tests carried over from #3274: two workers, lease lost mid-write, retries after takeover, crash at each step, person edits mid-run.
4. Config moved: old flags and config/deal_competitors.json become Signal Config rows; nothing else reads them.

18. Decisions made (Karthik, 2026-10-22) ↑ top

19. Still open ↑ top

Sources: SIGNALS-HARNESS-DESIGN-2026-10-22.md and SOURCES-DESIGN-2026-10-22.md (thr_2xt33ehbpp), Karthik's decisions 2026-10-22. ↑ back to top