diff --git a/.env.example b/.env.example index 0d6faae..9991a94 100644 --- a/.env.example +++ b/.env.example @@ -44,6 +44,16 @@ MODEL_NAME=google/gemini-2.5-flash # LAYA_SERVED_TIMEOUT_S=5 # per decision, the retry and any /health refresh included # LAYA_SERVED_MAX_LEN=512 # the served checkpoint's window: english 512, multilingual 1024 +# ---- Content desk (python -m s1a.content; draft Omni workers, not the shared --model flag) ---- +# Unset OMNI_BASE_URL and the desk uses the lexical rule. No Clef URL: that draft worker is refused. +# OMNI_BASE_URL=http://127.0.0.1:8000 +# OMNI_MODEL=StartLux-Decision-v1.1-4B-Instant +# OMNI_API_KEY= +# OMNI_TIMEOUT_S=30 +# OMNI_FACTCHECK_BASE_URL=http://127.0.0.1:8001 # required for --fact-check-worker pplx +# OMNI_FACTCHECK_MODEL=perplexity-ai/pplx-decider-v1.1-27b +# OMNI_FACTCHECK_API_KEY= + # ---- CLM (behind --model clm; `clm-serve` from the CLM checkout, no extra needed) ---- # CLM owns everything after a frozen Qwen3-8B encoder, so clm-serve needs an /v1/embeddings server beside it; # system1-omni's recipe/clm/README.md runs the pair. There is no cloud key. diff --git a/CHANGELOG.md b/CHANGELOG.md index 7ae591b..973903f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/); ver ### Added +- Content desk workflow (`python -m s1a.content`): RSS and fixture ingest, an optional X fixture or adapter + hook, a keep/drop screen with a relevance score, a human pick of 3–5 items, a fact-check choice and score, + and a human publish decision. Nothing is posted. `OMNI_BASE_URL` sends short `/v1/systemone` prefills to a + draft StartLux Instant 4B worker (system1-omni #164); fact-check can switch to draft PPLX (#166) on + `OMNI_FACTCHECK_BASE_URL`. Clef (#165) is refused. With the URL unset, a lexical rule runs offline. + `--allow-unsafe-skip-gates` is unsafe and still does not publish. `latency-report` is the unattached hook + for a later Public231-C1-style ceiling. Recipe: `recipes/content-workflow`. + - `s1a probe --order reverse|shuffle:N` (with `--seed` and `--json PATH`): the option-order check. Every case is asked in file order and again with the same keys and texts in another dict order; the probe reports the flip rate, accuracy in each order and a row per case with every order asked. Verdict and exit code are unchanged diff --git a/README.md b/README.md index afa6fe4..cf56f24 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,7 @@ The gates and the templates: [docs/skills.md](docs/skills.md#build-a-system-1-ag - `flights`: browser use, a Google Flights search over `@playwright/mcp`. - `desktop`: computer use, any Windows or macOS app window through [Cua Driver](https://cua.ai/docs/cua-driver). - `ticket_router`: 30 labelled support tickets to five queues. +- Content desk: RSS and fixture ingest, a screen, a human pick, a fact-check, and a human publish decision. Nothing is auto-published. Recipe: [recipes/content-workflow](recipes/content-workflow/README.md). - `alfworld`: household tasks in text, with the AI2-THOR scene in the replays. - `game2048`, `millionaire`, `blackjack`: games with a score per episode. - `injection_guard`: a rail that answers one question at a hook of a running agent and fails closed. diff --git a/docs/agents.md b/docs/agents.md index 75f36f2..a368b31 100644 --- a/docs/agents.md +++ b/docs/agents.md @@ -118,3 +118,15 @@ shipped set is 30 tickets in `s1a/agents/_data/ticket_router_eval.jsonl`, each w order status; the label is read by the scorer alone. `--dataset` points the agent at a JSONL of your own with the same fields, and `--batch-size` caps the tickets per episode. The rules text the model reads is `RULES` in `s1a/agents/ticket_router.py`. + +## The content desk + +The news desk is not a row in the table above. It is a five-stage command, `python -m s1a.content run`, because +the human pick and the human publish decision have to stop the run. The shared tool loop would act on every +choice. Screen and fact-check are still enumerated decisions: `keep`/`drop` plus a relevance score, then +`supported`/`unsupported`/`needs_human` plus a support score. With `OMNI_BASE_URL` unset, a lexical rule +answers, the same kind of baseline as the ticket router's keywords. With the variable set, a draft +System1-Omni worker answers over `POST /v1/systemone`. The default worker is StartLux Instant 4B. Fact-check +can switch to PPLX on `OMNI_FACTCHECK_BASE_URL`. Clef is refused. The run records a publish decision and +does not post. Details: [content-workflow.md](content-workflow.md) and the +[recipe](../recipes/content-workflow/README.md). diff --git a/docs/capabilities.md b/docs/capabilities.md index 7e882e0..adf8840 100644 --- a/docs/capabilities.md +++ b/docs/capabilities.md @@ -19,7 +19,7 @@ For issue trackers and serving support, see [System1-Omni model tracker #83](htt | Cua-S1 4B 0.2 text adapter (`Qwen/Qwen3.5-4B` + `cua-ai/cua-s1-4b-0.2/text`, `CUA_S1_VARIANT=4b`, `CUA_S1_MODALITY=text`) | Text | `choice` only | Tool decisions and text-only client paths; desktop option grounding is implemented | In-process Transformers/PEFT scorer | Contract tests use a fake scorer; real checkpoint/device validation **not present in this checkout**. | | Cua-S1 4B 0.2 multimodal adapter (`Qwen/Qwen3.5-4B` + `cua-ai/cua-s1-4b-0.2/multimodal`, variant `4b`) | One current screenshot plus text task/context | `choice` only; maximum 26 options in text and multimodal modes; multimodal requires exactly one screenshot | Screenshot-dependent desktop decision adapter; this does not itself complete the Omni capture/execute bridge | In-process Transformers/PEFT scorer with screenshot passed as a temporary file | Contract tests use a fake scorer; separate historical external evidence reports a bounded MPS fixture run (see below). Current-main reproduction and CUDA validation remain unverified. | -`random` is a seeded uniform-random choice baseline; `rule` is a hand-written one-hot choice policy. Neither is a model checkpoint. Both implement choice-only selection. `llm` refers to the separate chat model, not a decision-model backend. The `score` answer shape in Omni is not evidence that an Agents client can issue a score question. +`random` is a seeded uniform-random choice baseline; `rule` is a hand-written one-hot choice policy. Neither is a model checkpoint. Both implement choice-only selection. `llm` refers to the separate chat model, not a decision-model backend. The `score` answer shape in Omni is not evidence that an Agents client can issue a score question. The content desk (`python -m s1a.content`) is a separate client: when `OMNI_BASE_URL` is set it posts one short `choice` prefill and one short `score` prefill per item to a draft System1-Omni worker. That does not add `score` to `s1a decide`, and a live worker has not been measured here. See [content-workflow.md](content-workflow.md). ## Real validation evidence diff --git a/docs/configuration.md b/docs/configuration.md index 2490a5e..138942d 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -44,6 +44,13 @@ Variables can be exported in your shell or placed in a `.env` file at the root o | `LAYA_SERVED_API_KEY` | `laya-served` model | *(unset)* | Bearer token, the server's `LAYA_API_KEY` when it sets one. | | `LAYA_SERVED_TIMEOUT_S` | `laya-served` model | `5` | Deadline per decision in seconds, the one retry and any `/health` refresh included. A finite number above 0: `nan`, `inf` and `0` are configuration errors. | | `LAYA_SERVED_MAX_LEN` | `laya-served` model | `512` | The served checkpoint's token window per question (`english` 512, `multilingual` 1024); a request that fills it raises. The server takes the window from the checkpoint, so set this to match and nothing higher. | +| `OMNI_BASE_URL` | content desk, screen (and fact-check unless a second URL is set) | *(unset)* | Base URL of a draft System1-Omni worker, with no path or query. The client appends `/v1/systemone` and `/health`. Unset, `python -m s1a.content run --backend auto` uses the lexical rule and does not open a decision socket. | +| `OMNI_MODEL` | content desk screen | `StartLux-Decision-v1.1-4B-Instant` | Model alias sent to the screen worker. The default is the StartLux Instant 4B draft (system1-omni pull request 164). The thinking flag stays `think_mode: never`. | +| `OMNI_API_KEY` | content desk | *(unset)* | Bearer token for `OMNI_BASE_URL`, sent only when non-empty. | +| `OMNI_TIMEOUT_S` | content desk | `30` | Per-request timeout in seconds. A finite number above 0. There is no retry, so the latency hook can read one wall time. | +| `OMNI_FACTCHECK_BASE_URL` | content desk fact-check | *(same as `OMNI_BASE_URL` for StartLux)* | Required when `--fact-check-worker pplx`. PPLX (system1-omni pull request 166) is not assumed to share the StartLux port. | +| `OMNI_FACTCHECK_MODEL` | content desk fact-check | StartLux alias, or `perplexity-ai/pplx-decider-v1.1-27b` for PPLX | Model alias for the fact-check worker. | +| `OMNI_FACTCHECK_API_KEY` | content desk fact-check | `OMNI_API_KEY` | Bearer token for the fact-check URL when it differs. | | `CUA_S1_CHECKPOINT` | `cua` model | `cua-ai/cua-s1-nano-0.1` | Hugging Face checkpoint ID or local directory for Cua-S1 Nano option scorer. | | `CUA_S1_SUBFOLDER` | `cua` model | `text` | Subfolder within checkpoint directory containing text option scoring weights. | | `CUA_S1_DEVICE` | `cua` model | `auto` | PyTorch device used for Cua-S1 Nano evaluation (`auto`, `cpu`, `cuda`, or `mps`). | diff --git a/docs/content-workflow.md b/docs/content-workflow.md new file mode 100644 index 0000000..cb9d02b --- /dev/null +++ b/docs/content-workflow.md @@ -0,0 +1,91 @@ +# Content workflow + +Jev screens. A person decides. The desk path is five stages: + +1. Ingest news from RSS and, when an adapter is supplied, Twitter/X. +2. Screen each item to keep or drop, and score relevance. +3. A person picks 3 to 5 kept items for deep research. +4. Fact-check those items' claims against the sources the person supplied, as a choice and a score. +5. A person decides publish or hold. The run records that decision and posts nothing. + +System1-Agents owns ingest, the gates, and the report. A draft System1-Omni worker is used only for stage 2 and stage 4, and only when `OMNI_BASE_URL` is set. The shared tool loop is not used: that loop would act on every choice, and these two gates must be able to stop the run. + +The diagram that asks for about 200 items down to 30–50, then 3–5, is the shape of a live desk. The shipped fixture is 50 labelled items and keeps 32. It does not measure the diagram's 15 seconds or ¥0.01. + +## Where each stage lives + +| Stage | Code | What it refuses to do | +|---|---|---| +| Ingest RSS | `load_rss_file` / `load_rss_url` in `s1a/content/ingest.py` | A path is a file. A URL is opt-in via `--rss-url`. | +| Ingest X | `load_x_fixture`, `posts_to_items`, `UnconfiguredXAdapter` | No request is sent to `api.x.com`. `X_BEARER_TOKEN` is not read. | +| Screen | `RuleDecider.screen` or `OmniDecider.screen` | Labels and drafts are not copied into the prefill. | +| Human pick | `run_workflow` in `s1a/content/workflow.py` | Missing picks raise `GateRequired` before fact-check. | +| Fact-check | `RuleDecider.fact_check` or `OmniDecider.fact_check` | No draft means `needs_human` and no model call. The worker does not write the research. | +| Human publish | the same `run_workflow` | `human_publish.executed` is always `[]`. `auto_publish` is always false. | + +A live X adapter is a `fetch()` callable passed to `load_inputs` as `x_fetch`. It should `GET https://api.x.com/2/tweets/search/recent` with `Authorization: Bearer $X_BEARER_TOKEN`, `query`, and `max_results`, then return posts that `posts_to_items` can map. That call is the operator's code. `UnconfiguredXAdapter.fetch` raises instead. + +## Decision shape + +Screen choices are `keep` and `drop`. The relevance score uses two levels, "Not relevant" then "Relevant hotspot". Fact-check choices are `supported`, `unsupported`, and `needs_human`. The support score uses three levels, "Unsupported", "Needs a human", then "Supported". + +Those are the same kind of enumerated questions as `ticket_router` and `s1a decide`. `s1a decide` still answers one `choice`. It does not grow a `score` question. Score is asked only by `s1a/content/omni_client.py`, one short prefill per question, Public231-style: a string `state`, one entry in `questions`, and no images. + +The offline decider is a lexical rule, the same role as the ticket router's keyword baseline. On the shipped fixture it matches the labels by construction. That is a wiring check, not a Jev or Omni quality result. The rule keeps a whole word such as `model` or `agent`. It also keeps "model trains", and it drops "supermodel", "chipotle", and "agency". The fixture simply does not include the "model trains" trap. + +## Draft workers + +Nothing in pull requests 164–166 has to be merged. The client needs a base URL that already speaks `POST /v1/systemone` and `GET /health`. + +| Role | Worker | Draft PR | Request flag | +|---|---|---|---| +| Screen, and fact-check by default | StartLux Instant 4B (`StartLux-Decision-v1.1-4B-Instant`) | [system1-omni #164](https://github.com/InferMatrixLab/system1-omni/pull/164) | `think_mode: never`. The `thinking` object is not sent. | +| Optional heavier fact-check | PPLX decider (`perplexity-ai/pplx-decider-v1.1-27b`) | [system1-omni #166](https://github.com/InferMatrixLab/system1-omni/pull/166) | `thinking: {"enabled": false}` on `OMNI_FACTCHECK_BASE_URL`. | +| Not on this path | Clef / Clef-flash | [system1-omni #165](https://github.com/InferMatrixLab/system1-omni/pull/165) | Refused. | + +`OMNI_MODEL` overrides the StartLux alias. `OMNI_FACTCHECK_MODEL` overrides the fact-check alias. PPLX is not assumed to share StartLux's port: `--fact-check-worker pplx` without `OMNI_FACTCHECK_BASE_URL` is an error. Clef is an error on either step. + +With `OMNI_BASE_URL` unset, `--backend auto` is the lexical rule and opens no decision socket. `--backend omni` without that variable is a usage error. + +## Human gates + +Pass `--picks` (`{"ids": [...]}`) and `--publish` (`{"decisions": {"id": "publish"|"hold"}}`). On a terminal, omitting them prompts. Otherwise the process exits 2 and writes a packet only when `--packet` is set. The packet's `stopped_at` is `human_pick` or `human_publish`. It is not a finished run. + +`--allow-unsafe-skip-gates` is unsafe. It invents a pick (the first three kept ids, or every kept id when fewer than three were kept) and holds every picked item. It does not publish. An explicit file or hook wins over the flag: the flag does not replace a decision you already passed. + +A decision of `publish` is still only a record. There is no publisher, no CMS write, and no MCP tool that posts. Hosts that speak MCP should collect the human files and run this command. `run_agent` is the wrong tool: it finishes a loop, and this desk has to stop for a person. + +Python hooks, for a caller that already has the person in the loop: + +```python +from s1a.content.rules import RuleDecider +from s1a.content.workflow import run_workflow + +run_workflow(items, decider=RuleDecider(), pick_hook=ask_for_ids, publish_hook=ask_publish_or_hold) +``` + +`pick_hook` sees kept items without labels. `publish_hook` sees the fact-check choice and score name. Both results are checked: 3 to 5 kept ids when at least three were kept, otherwise every kept id; publish decisions must name each picked id as `publish` or `hold`. + +## Latency gate, not attached + +Every report carries `latency_gate.attached: false`. Per-request client wall times are stored on Omni calls (`calls[].latency_ms`). Rule rows are omitted from the gate so a 0 ms baseline cannot clear a ceiling. + +Hang a Public231-C1-style ceiling later, after a worker exists: + +1. Serve the draft worker. Confirm `GET $OMNI_BASE_URL/health`. +2. Discard a warmup of 5 requests. The workflow itself does not drop them; run a throwaway command first, or strip those calls from the report. +3. Keep concurrency at 1. This client already sends one POST at a time. +4. Repeat three measured rounds on the same fixture. +5. Pool the successful request latencies and run: + +```bash +python -m s1a.content latency-report \ + --report runs/recipes/content-workflow/report.json \ + --p50-max-ms \ + --p95-max-ms \ + --min-samples +``` + +No ceiling is pinned. A p50 from Public231 on another checkpoint is not this fixture's budget. The clock here is `client_request_wall_ms` (POST start through the response body). System1-Omni's bench clock is `post_to_body_json_and_schema_validation`. Do not treat them as one number. With a ceiling set and too few Omni samples, the command fails closed and exits 1. With no ceiling, it prints the summary and exits 0. + +The runnable fixture path is [the recipe](../recipes/content-workflow/README.md). diff --git a/docs/skills.md b/docs/skills.md index bebb4a9..f39ac65 100644 --- a/docs/skills.md +++ b/docs/skills.md @@ -58,6 +58,9 @@ runs `s1a run flights --model jev --goal "..."` and reads the answer from `final - Keep arithmetic, constraint deduction, search over move trees, free-text generation and any value the page never shows with the chat model. The three probe misses, Minesweeper, Wordle and a severity rubric, needed deduction. A plain page fetch needs no agent; use `curl`. +- A news desk is `python -m s1a.content run`, not a chat draft and not `run_agent`. Jev or a draft Omni worker + screens and fact-checks. A person picks 3 to 5 items and decides publish or hold. The command does not post. + See [content-workflow.md](content-workflow.md). ### What comes back diff --git a/recipes/README.md b/recipes/README.md index 70ee100..012a8fa 100644 --- a/recipes/README.md +++ b/recipes/README.md @@ -10,6 +10,7 @@ backend, and the required application + System1-Agents + System1-Omni video for | Use case | Agent / front | Execution profile | Validation | |---|---|---|---| | [Route support tickets](ticket-routing/README.md) | `ticket_router` / tool | Local CPU rule baseline; Laya/Jev alternatives, System1-Omni served Laya, or CLM over HTTP | Baseline smoke run; this fixture's required served-model video remains pending, and CLM measured 18/90 under this loop's framing ([evidence](../evals/ticket_router/CLM.md)) | +| [Screen a news desk](content-workflow/README.md) | `s1a.content` orchestration, not a tool-loop agent | Local CPU lexical rule; optional draft StartLux/PPLX `/v1/systemone` workers | Fixture smoke run with human pick and publish files; nothing is posted. Served-worker video remains pending | | [Play Snake](snake/README.md) | snake game client (`evals/snake`) / tool-style loop | System1-Omni native worker (H800); MPS dev path | 3×2400-step (0 deaths) and 16×600-step multigrid (0 deaths) recorded runs, playback-1 GIFs | The recipe's validation section identifies what was actually run. A supported model or a passing unit test diff --git a/recipes/content-workflow/README.md b/recipes/content-workflow/README.md new file mode 100644 index 0000000..4cdbdd2 --- /dev/null +++ b/recipes/content-workflow/README.md @@ -0,0 +1,199 @@ +# Screen a news desk, then wait for a person + +## Summary + +Run the five-stage desk in `s1a.content`: ingest, screen, human pick, fact-check, human publish decision. +The observable check is the fixture report: 50 labelled items screened, 4 picked items fact-checked, the +publish decisions equal to `publish.json`, and `human_publish.executed` empty. A finished JSON object with +`auto_publish: true` or a non-empty `executed` list is a failure even when the process exits 0. + +The offline profile is a lexical rule, the same kind of baseline as `ticket_router`'s keywords. It matches +this fixture's labels by construction. That checks the wiring. It is not a quality claim for Jev, StartLux, +or PPLX. Deep research is the drafts already stored on the picked items. The run does not write them. + +| Profile | Inference path | Requirements / validation | +|---|---|---| +| `--backend rule` (also `--backend auto` with `OMNI_BASE_URL` unset) | Lexical keep/drop and fact rule in-process | Core install, CPU, no model and no decision socket. Recorded below. | +| `--backend omni` | StartLux Instant 4B draft worker, `POST /v1/systemone`, short prefill | `OMNI_BASE_URL`. Draft [system1-omni #164](https://github.com/InferMatrixLab/system1-omni/pull/164) does not need to be merged. Command not run here. | +| `--backend omni --fact-check-worker pplx` | Screen stays StartLux; fact-check uses the PPLX draft worker | `OMNI_FACTCHECK_BASE_URL`. Draft [system1-omni #166](https://github.com/InferMatrixLab/system1-omni/pull/166). Command not run here. | +| Clef | Refused | [system1-omni #165](https://github.com/InferMatrixLab/system1-omni/pull/165) is not on this hot path. | + +## Prerequisites and setup + +From a System1-Agents checkout, Python 3.11+ and `uv`: + +```bash +uv sync --frozen +uv run python -m s1a.content run --help +mkdir -p runs/recipes/content-workflow +git rev-parse HEAD +``` + +Keep the printed revision with the result. The core install is enough. No browser, GPU, model download, or +API key is required for the rule profile. `MODEL_NAME` is unused: this path does not call the chat model. + +The labelled sample is `items.jsonl` (50 items, 32 `keep`, 18 `drop`, 8 drafts). `picks.json` names four of +the kept drafts. `publish.json` records the human decision, including `publish` for an item the fact-check +calls unsupported. `sample.rss` and `x_posts.jsonl` are the offline RSS and X adapters. URLs in the sample +use `example.invalid` and are not fetched. + +File sha256 of `items.jsonl` at the recorded run: `51828b6671f507e4ce635d90a43420b1855fde864c43dfc17c072e3ba983f064`. + +## Run the task + +```bash +OMNI_BASE_URL= OMNI_API_KEY= uv run python -m s1a.content run \ + --items recipes/content-workflow/items.jsonl \ + --picks recipes/content-workflow/picks.json \ + --publish recipes/content-workflow/publish.json \ + --backend rule \ + > runs/recipes/content-workflow/report.json +``` + +Stdout is one JSON object. A missing gate prints one line on stderr and exits 2. `--backend rule` does not +read `OMNI_BASE_URL`. The shell prefix clears it so a developer environment cannot switch this command onto +a worker. + +## Verify the result + +The check uses the fixture labels, the pick file, and the publish file. It fails on a short screen, a wrong +choice or score, a skipped gate, or anything that looks like a post: + +```bash +uv run python - <<'PY' +import json +from pathlib import Path + +report = json.loads(Path("runs/recipes/content-workflow/report.json").read_text()) +items = [json.loads(line) for line in Path("recipes/content-workflow/items.jsonl").read_text().splitlines() if line.strip()] +picks = json.loads(Path("recipes/content-workflow/picks.json").read_text())["ids"] +publish = json.loads(Path("recipes/content-workflow/publish.json").read_text())["decisions"] +labels = {row["id"]: row["label"] for row in items} +fact_labels = {row["id"]: row["draft"]["label"] for row in items if row.get("draft")} +screen = {row["id"]: row for row in report["screen"]["records"]} +assert report["backend"] == "rule" and report["unsafe_skip"] is False +assert report["stages"] == ["ingest", "screen", "human_pick", "fact_check", "human_publish"] +assert report["screen"]["total"] == report["screen"]["labeled"] == 50 +assert report["screen"]["kept"] == 32 +assert report["screen"]["choice_accuracy"] == report["screen"]["score_accuracy"] == 1.0 +assert report["ingest"]["sources"]["rss"] > 0 and report["ingest"]["sources"]["x"] > 0 +assert sum(report["ingest"]["sources"].values()) == 50 +for ident, label in labels.items(): + assert screen[ident]["choice"] == label + assert screen[ident]["choice_correct"] and screen[ident]["score_correct"] +assert report["human_pick"]["ids"] == picks +facts = report["fact_check"]["records"] +assert [row["id"] for row in facts] == picks +for row in facts: + assert row["choice"] == fact_labels[row["id"]] + assert row["choice_correct"] and row["score_correct"] + print(row["id"], "->", row["choice"], "score", row["score_level"], "decision", publish[row["id"]]) +assert report["human_publish"]["decisions"] == publish +assert "publish" in publish.values() +assert report["human_publish"]["executed"] == [] and report["auto_publish"] is False +assert report["latency_gate"]["attached"] is False +print("screen", report["screen"]["choice_correct"], "/", report["screen"]["total"]) +print("Verified the fixture workflow; nothing was published") +PY +``` + +The expected closing line is `Verified the fixture workflow; nothing was published`. Exit 0 without that +line is not success. A stopped gate keeps a partial packet only when `--packet` is set, and that packet is +not this report. + +Recorded rule-profile result, parent revision `9b20b989d919bc69227f81bc0ec571ca56925586`, before the commit +that added this recipe: screen 50/50, fact-check `n01 supported`, `n02 unsupported`, `n03 needs_human`, +`n04 supported`, publish decisions unchanged, `executed` empty. The same command without `--picks` exited 2 +with `human pick gate` on stderr. Details: [smoke-evidence.json](smoke-evidence.json). + +## Use a draft Omni worker + +Do not point this path at Clef. StartLux is the default screen and fact-check worker. PPLX is an optional +second server for fact-check only. Neither draft pull request has to be merged; the process only needs the +URL. No live worker was running for the recorded run, so the blocks below are unchecked commands. + +Readiness, then one screened run. `OMNI_API_KEY` is sent as a bearer token only when it is non-empty. + +```bash +curl --fail "$OMNI_BASE_URL/health" +uv run python -m s1a.content run \ + --items recipes/content-workflow/items.jsonl \ + --picks recipes/content-workflow/picks.json \ + --publish recipes/content-workflow/publish.json \ + --backend omni \ + > runs/recipes/content-workflow/report.json +``` + +The report's `workers.screen` names the model, the draft pull request, and `prefill: short`. Each screen +record's `calls` has two request ids, `screen` and `relevance`. Agreement with the lexical labels is a +comparison, not a pass bar: a worker that disagrees is still a successful desk run if the gates held and +nothing was published. Re-run the verifier's structural checks, and do not require `choice_accuracy == 1` +for a worker. Keep the worker's misses in the report. + +Heavier fact-check on a second URL: + +```bash +uv run python -m s1a.content latency-report --report runs/recipes/content-workflow/report.json +# after three measured rounds, warmup excluded, and only if you have a ceiling: +# uv run python -m s1a.content latency-report --report report.json --p95-max-ms --min-samples +``` + +```bash +OMNI_FACTCHECK_BASE_URL=http://127.0.0.1:8001 uv run python -m s1a.content run \ + --items recipes/content-workflow/items.jsonl \ + --picks recipes/content-workflow/picks.json \ + --publish recipes/content-workflow/publish.json \ + --backend omni --fact-check-worker pplx \ + > runs/recipes/content-workflow/report.json +``` + +`latency-report` with a ceiling and no Omni samples fails closed (exit 1). That is the hook for a later +Public231 C1 gate: warmup 5 excluded, concurrency 1, three rounds, then p50/p95. The clock is +`client_request_wall_ms`, not Omni's `post_to_body_json_and_schema_validation`. No ceiling is pinned. See +[the workflow doc](../../docs/content-workflow.md#latency-gate-not-attached). + +One item can also be asked as a choice through `s1a decide`. That command has no score head, needs a Jev +key, and was not run for this recipe: + +```bash +uv run s1a decide \ + --state '{"title": "Open model checkpoint lands for local agents", "summary": "A small checkpoint for agent inference is up for local tests.", "source": "rss"}' \ + --option keep="Relevant AI, model, agent, inference, chip, checkpoint or research hotspot" \ + --option drop="Not a relevant hotspot for this desk" \ + --rules "Keep a relevant hotspot for an AI-agent news desk. Drop sports, weather, recipes, traffic, celebrity and anything without that hook." +``` + +## Demo and validation + +The required application + System1-Agents + System1-Omni video is not in this recipe. No draft worker was +served here, and pull requests 164 and 166 are unmerged. A rule-backend terminal run does not fill that +gap. The recipe stays an evidence gap for that video until a maintainer accepts the exception or a later +run supplies the clip: input items, the worker's choice and score, the human files, `executed: []`, and +the answering model, request id, and base URL from the same run. + +`tests/test_content_workflow.py` covers the fixture, both gates, the unsafe flag, RSS and X fixtures, and +a mock StartLux/PPLX transport. It does not open a socket except the CLI test that expects a refused +connection at `127.0.0.1:9`. + +## Troubleshooting and limits + +| Symptom | Check | +|---|---| +| Exit 2, `human pick gate` | Pass `--picks`, answer the prompt on a terminal, or read the unsafe flag's help before using it | +| Exit 2, `human publish gate` | The stderr line lists fact-check choices. Pass `--publish`. Re-running repeats the screen | +| Exit 2, pick was not kept | The id was dropped. Pick from `screen.kept_ids` | +| `PPLX fact-check needs OMNI_FACTCHECK_BASE_URL` | PPLX is a different draft server. Do not send it to the StartLux port by default | +| `Clef is not on the content hot path` | Use StartLux, or PPLX for fact-check only | +| Omni HTTP error | The message has the status, not the body. Check `/health` and the model alias | +| `choice_accuracy` below 1 on `omni` | Expected until a worker is measured. The rule profile is the wiring check | + +The lexical rule is brittle: the word "model" in "model trains" would be kept. This fixture does not +contain that sentence. Fifty synthetic items are not a desk accuracy claim. RSS URLs and X are not +contacted by the recorded command. A human `publish` value is not a post. + +## References + +- [Workflow doc](../../docs/content-workflow.md) and [configuration](../../docs/configuration.md) +- [Ticket-routing recipe](../ticket-routing/README.md), the batch-decision analogue +- [Decision-model interface](../../docs/decision-models.md) +- Draft workers: [StartLux #164](https://github.com/InferMatrixLab/system1-omni/pull/164), [PPLX #166](https://github.com/InferMatrixLab/system1-omni/pull/166). Clef [#165](https://github.com/InferMatrixLab/system1-omni/pull/165) is excluded. diff --git a/recipes/content-workflow/items.jsonl b/recipes/content-workflow/items.jsonl new file mode 100644 index 0000000..3f759ce --- /dev/null +++ b/recipes/content-workflow/items.jsonl @@ -0,0 +1,50 @@ +{"id": "n01", "title": "Open model checkpoint lands for local agents", "summary": "A small checkpoint for agent inference is up for local tests.", "source": "rss", "url": "https://example.invalid/n01", "label": "keep", "draft": {"claims": ["StartLux Instant 4B is the default demo worker for this desk."], "sources": [{"id": "s1", "text": "The default demo worker for this desk is StartLux Instant 4B."}], "label": "supported"}} +{"id": "n02", "title": "Research note on agent screen costs", "summary": "The note compares a keep or drop screen with a later fact check.", "source": "rss", "url": "https://example.invalid/n02", "label": "keep", "draft": {"claims": ["The screen finished in 15 seconds."], "sources": [{"id": "s1", "text": "The screen is a keep or drop choice."}], "label": "unsupported"}} +{"id": "n03", "title": "Desk research leaves a thin item", "summary": "The item has a title and almost no source text.", "source": "rss", "url": "https://example.invalid/n03", "label": "keep", "draft": {"claims": ["Worth a closer look later."], "sources": [{"id": "s1", "text": "A short desk note."}], "label": "needs_human"}} +{"id": "n04", "title": "Human reviewers and the agent desk", "summary": "Reviewers pick a few agent items before anyone publishes.", "source": "rss", "url": "https://example.invalid/n04", "label": "keep", "draft": {"claims": ["Human reviewers pick three to five items before fact checking."], "sources": [{"id": "s1", "text": "Human reviewers pick three to five items before fact checking and then decide."}], "label": "supported"}} +{"id": "n05", "title": "Local inference on a laptop chip", "summary": "A lab note says local inference can run on a laptop chip.", "source": "x", "url": "https://example.invalid/n05", "label": "keep", "draft": {"claims": ["Local inference can run on a laptop chip."], "sources": [{"id": "s1", "text": "Local inference can run on a laptop chip in the lab."}], "label": "supported"}} +{"id": "n06", "title": "New checkpoint report overstates size", "summary": "The report talks about a checkpoint and a model.", "source": "x", "url": "https://example.invalid/n06", "label": "keep", "draft": {"claims": ["The checkpoint has 27 billion parameters."], "sources": [{"id": "s1", "text": "The checkpoint is an Instant 4B model."}], "label": "unsupported"}} +{"id": "n07", "title": "Agent weekly roundup stays vague", "summary": "The roundup lists an agent item without a source quote.", "source": "rss", "url": "https://example.invalid/n07", "label": "keep", "draft": {"claims": ["Maybe relevant."], "sources": [{"id": "s1", "text": "See the notes."}], "label": "needs_human"}} +{"id": "n08", "title": "Publish policy for model notes", "summary": "The policy says publish stays a human decision.", "source": "rss", "url": "https://example.invalid/n08", "label": "keep", "draft": {"claims": ["Publish stays a human decision."], "sources": [{"id": "s1", "text": "Publish stays a human decision and the tool does not post."}], "label": "supported"}} +{"id": "n09", "title": "Chip roadmap mentions on-device inference", "summary": "Vendors describe a chip built for inference.", "source": "rss", "url": "https://example.invalid/n09", "label": "keep"} +{"id": "n10", "title": "LLM eval roundup from the lab", "summary": "The lab posts a short LLM eval note.", "source": "rss", "url": "https://example.invalid/n10", "label": "keep"} +{"id": "n11", "title": "Open research preprint on agent memory", "summary": "A research preprint looks at agent memory.", "source": "rss", "url": "https://example.invalid/n11", "label": "keep"} +{"id": "n12", "title": "Checkpoint release for a small model", "summary": "The checkpoint is a small model aimed at laptops.", "source": "rss", "url": "https://example.invalid/n12", "label": "keep"} +{"id": "n13", "title": "Agent framework adds a tool hook", "summary": "The framework is an agent tool hook, not a chat product.", "source": "rss", "url": "https://example.invalid/n13", "label": "keep"} +{"id": "n14", "title": "Inference server trims batch latency", "summary": "The inference server note is about batching.", "source": "rss", "url": "https://example.invalid/n14", "label": "keep"} +{"id": "n15", "title": "Research group shares a model card", "summary": "The model card is a research note.", "source": "rss", "url": "https://example.invalid/n15", "label": "keep"} +{"id": "n16", "title": "New chip brief for local LLMs", "summary": "The brief is about a chip for local LLMs.", "source": "rss", "url": "https://example.invalid/n16", "label": "keep"} +{"id": "n17", "title": "Agent desk tracks checkpoint diffs", "summary": "The desk tracks checkpoint diffs for an agent.", "source": "rss", "url": "https://example.invalid/n17", "label": "keep"} +{"id": "n18", "title": "Model lab publishes an inference guide", "summary": "The guide is about model inference.", "source": "rss", "url": "https://example.invalid/n18", "label": "keep"} +{"id": "n19", "title": "Research recap on agent routing", "summary": "A research recap covers agent routing.", "source": "rss", "url": "https://example.invalid/n19", "label": "keep"} +{"id": "n20", "title": "LLM checkpoint mirrored for offline tests", "summary": "An LLM checkpoint is mirrored for offline tests.", "source": "rss", "url": "https://example.invalid/n20", "label": "keep"} +{"id": "n21", "title": "Chip vendor demo of agent inference", "summary": "A chip vendor demos agent inference.", "source": "x", "url": "https://example.invalid/n21", "label": "keep"} +{"id": "n22", "title": "Small model fine-tune for research tasks", "summary": "A small model fine-tune targets research tasks.", "source": "x", "url": "https://example.invalid/n22", "label": "keep"} +{"id": "n23", "title": "Agent memory paper draft circulates", "summary": "An agent memory paper is a research draft.", "source": "x", "url": "https://example.invalid/n23", "label": "keep"} +{"id": "n24", "title": "Inference stack update for the checkpoint", "summary": "The inference stack update names the checkpoint.", "source": "x", "url": "https://example.invalid/n24", "label": "keep"} +{"id": "n25", "title": "Lab posts an LLM screening rubric", "summary": "The lab posts an LLM screening rubric.", "source": "x", "url": "https://example.invalid/n25", "label": "keep"} +{"id": "n26", "title": "Open model weights for an agent demo", "summary": "Open model weights are for an agent demo.", "source": "x", "url": "https://example.invalid/n26", "label": "keep"} +{"id": "n27", "title": "Research seminar on inference chips", "summary": "A research seminar covers inference chips.", "source": "x", "url": "https://example.invalid/n27", "label": "keep"} +{"id": "n28", "title": "Checkpoint hash pinned in the model note", "summary": "The model note pins a checkpoint hash.", "source": "x", "url": "https://example.invalid/n28", "label": "keep"} +{"id": "n29", "title": "Agent eval uses a local LLM", "summary": "The agent eval uses a local LLM.", "source": "x", "url": "https://example.invalid/n29", "label": "keep"} +{"id": "n30", "title": "Chip brief compares two inference paths", "summary": "The chip brief compares two inference paths.", "source": "x", "url": "https://example.invalid/n30", "label": "keep"} +{"id": "n31", "title": "Model card lists research limits", "summary": "The model card lists research limits.", "source": "x", "url": "https://example.invalid/n31", "label": "keep"} +{"id": "n32", "title": "LLM agent recipe stays on fixtures", "summary": "The LLM agent recipe stays on fixtures.", "source": "x", "url": "https://example.invalid/n32", "label": "keep"} +{"id": "n33", "title": "Supermodel opens a downtown cafe", "summary": "The opening is downtown on Saturday.", "source": "rss", "url": "https://example.invalid/n33", "label": "drop"} +{"id": "n34", "title": "Cafe adds chipotle sauce", "summary": "The special is on the lunch board.", "source": "rss", "url": "https://example.invalid/n34", "label": "drop"} +{"id": "n35", "title": "Agency posts holiday hours", "summary": "The office closes Friday.", "source": "rss", "url": "https://example.invalid/n35", "label": "drop"} +{"id": "n36", "title": "City marathon road closures", "summary": "Downtown streets close for the Sunday race.", "source": "rss", "url": "https://example.invalid/n36", "label": "drop"} +{"id": "n37", "title": "Rain expected overnight", "summary": "Showers move in after midnight.", "source": "rss", "url": "https://example.invalid/n37", "label": "drop"} +{"id": "n38", "title": "Ferry timetable changes Sunday", "summary": "The morning ferry leaves twenty minutes later.", "source": "rss", "url": "https://example.invalid/n38", "label": "drop"} +{"id": "n39", "title": "Library sale this weekend", "summary": "Paperbacks are stacked in the lobby.", "source": "rss", "url": "https://example.invalid/n39", "label": "drop"} +{"id": "n40", "title": "Bakery sells out of buns", "summary": "The morning batch went before noon.", "source": "rss", "url": "https://example.invalid/n40", "label": "drop"} +{"id": "n41", "title": "Park concert draws a crowd", "summary": "The band plays until ten.", "source": "x", "url": "https://example.invalid/n41", "label": "drop"} +{"id": "n42", "title": "School play opens Thursday", "summary": "Tickets are at the door.", "source": "x", "url": "https://example.invalid/n42", "label": "drop"} +{"id": "n43", "title": "Harbor fog delays the tour", "summary": "Boats wait at the pier.", "source": "x", "url": "https://example.invalid/n43", "label": "drop"} +{"id": "n44", "title": "Farmers market moves indoors", "summary": "Stalls line the hall.", "source": "x", "url": "https://example.invalid/n44", "label": "drop"} +{"id": "n45", "title": "Weekend crossword answers posted", "summary": "The puzzle editor shares the grid.", "source": "x", "url": "https://example.invalid/n45", "label": "drop"} +{"id": "n46", "title": "Train museum opens a new hall", "summary": "Old cars sit on a side track.", "source": "x", "url": "https://example.invalid/n46", "label": "drop"} +{"id": "n47", "title": "Quilt fair fills the gym", "summary": "Booths open at nine.", "source": "x", "url": "https://example.invalid/n47", "label": "drop"} +{"id": "n48", "title": "River path closed for repairs", "summary": "Walkers detour by the bridge.", "source": "rss", "url": "https://example.invalid/n48", "label": "drop"} +{"id": "n49", "title": "Choir rehearsal moves to Tuesday", "summary": "The group meets in the hall.", "source": "rss", "url": "https://example.invalid/n49", "label": "drop"} +{"id": "n50", "title": "Night market returns Friday", "summary": "Stalls sell fruit and tea.", "source": "rss", "url": "https://example.invalid/n50", "label": "drop"} diff --git a/recipes/content-workflow/picks.json b/recipes/content-workflow/picks.json new file mode 100644 index 0000000..5ffa32d --- /dev/null +++ b/recipes/content-workflow/picks.json @@ -0,0 +1,3 @@ +{ + "ids": ["n01", "n02", "n03", "n04"] +} diff --git a/recipes/content-workflow/publish.json b/recipes/content-workflow/publish.json new file mode 100644 index 0000000..c21846f --- /dev/null +++ b/recipes/content-workflow/publish.json @@ -0,0 +1,8 @@ +{ + "decisions": { + "n01": "publish", + "n02": "publish", + "n03": "hold", + "n04": "hold" + } +} diff --git a/recipes/content-workflow/sample.rss b/recipes/content-workflow/sample.rss new file mode 100644 index 0000000..ffdbbc9 --- /dev/null +++ b/recipes/content-workflow/sample.rss @@ -0,0 +1,18 @@ + + + + Desk fixture + + New agent checkpoint for local inference + https://example.invalid/rss-keep-1 + rss-keep-1 + A small checkpoint for agent inference. + + + City marathon road closures + https://example.invalid/rss-drop-1 + rss-drop-1 + Downtown streets close for the Sunday race. + + + diff --git a/recipes/content-workflow/smoke-evidence.json b/recipes/content-workflow/smoke-evidence.json new file mode 100644 index 0000000..c8f8cc4 --- /dev/null +++ b/recipes/content-workflow/smoke-evidence.json @@ -0,0 +1,44 @@ +{ + "recorded_at": "2026-10-25T02:08:22Z", + "base_revision": "9b20b989d919bc69227f81bc0ec571ca56925586", + "working_tree_scope": "Content workflow, recipe, and tests recorded in the working tree that adds them, on top of the base revision. Re-run the command on the PR head.", + "items_sha256": "51828b6671f507e4ce635d90a43420b1855fde864c43dfc17c072e3ba983f064", + "report_items_sha256": "312628ea432a06e5590334c5625663c01cf61f34e3d110cba4be0f2f3ffadec2", + "environment": { + "platform": "Linux x86_64", + "python": "Python 3.12.3", + "execution": "CPU lexical rule; no neural inference and no decision HTTP call" + }, + "command": "OMNI_BASE_URL= OMNI_API_KEY= uv run python -m s1a.content run --items recipes/content-workflow/items.jsonl --picks recipes/content-workflow/picks.json --publish recipes/content-workflow/publish.json --backend rule", + "sanitization": "The full report stays under runs/, which is gitignored. Counts below are from that report.", + "verification_output": [ + "n01 -> supported score 2 decision publish", + "n02 -> unsupported score 0 decision publish", + "n03 -> needs_human score 1 decision hold", + "n04 -> supported score 2 decision hold", + "screen 50 / 50", + "Verified the fixture workflow; nothing was published" + ], + "summary": { + "backend": "rule", + "screen_total": 50, + "screen_kept": 32, + "screen_choice_correct": 50, + "screen_score_correct": 50, + "fact_choice_correct": 4, + "fact_score_correct": 4, + "publish_decisions": {"n01": "publish", "n02": "publish", "n03": "hold", "n04": "hold"}, + "executed": [], + "auto_publish": false, + "unsafe_skip": false, + "latency_gate_attached": false, + "sources": {"rss": 29, "x": 21, "fixture": 0} + }, + "negative_verifier_checks": [ + { + "input": "run without --picks", + "exit_code": 2, + "stderr": "human pick gate: pass picks, answer the prompt, or pass --allow-unsafe-skip-gates (unsafe: it invents a pick and holds every item; it still does not publish)" + } + ] +} diff --git a/recipes/content-workflow/x_posts.jsonl b/recipes/content-workflow/x_posts.jsonl new file mode 100644 index 0000000..1cdd619 --- /dev/null +++ b/recipes/content-workflow/x_posts.jsonl @@ -0,0 +1,2 @@ +{"id": "x-local-1", "text": "Research note on agent inference chips.", "author": "desk"} +{"id": "x-local-2", "text": "Rain expected overnight.", "author": "weather"} diff --git a/s1a/content/__init__.py b/s1a/content/__init__.py new file mode 100644 index 0000000..da18f53 --- /dev/null +++ b/s1a/content/__init__.py @@ -0,0 +1,11 @@ +# coding: utf-8 +"""Desk workflow: ingest, screen, human pick, fact-check, human publish decision. + +System1-Agents owns this orchestration. A draft System1-Omni worker is used only when +``OMNI_BASE_URL`` is set, and only for the screen and fact-check questions. The package +never posts a story. +""" + +from s1a.content.workflow import GateRequired, run_workflow + +__all__ = ["GateRequired", "run_workflow"] diff --git a/s1a/content/__main__.py b/s1a/content/__main__.py new file mode 100644 index 0000000..dbd955b --- /dev/null +++ b/s1a/content/__main__.py @@ -0,0 +1,213 @@ +# coding: utf-8 +"""``python -m s1a.content``: the five-stage desk run, and the later latency-gate command. + +Stdout is one JSON object on success. A usage or gate error is one line on stderr and exit 2. +An Omni or feed failure is one line on stderr and exit 1. A completed run never publishes. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path +from typing import Any + +import httpx +from dotenv import load_dotenv + +from s1a.content.ingest import load_inputs +from s1a.content.latency import gate, omni_call_latencies +from s1a.content.omni_client import decider_from_env, require_worker, worker_record +from s1a.content.rules import RuleDecider +from s1a.content.workflow import GateRequired, run_workflow + +ROOT = Path(__file__).resolve().parents[2] + + +def parser() -> argparse.ArgumentParser: + build = argparse.ArgumentParser(prog="python -m s1a.content", description=__doc__) + commands = build.add_subparsers(dest="command", required=True) + run = commands.add_parser("run", help="ingest, screen, pick, fact-check, and record a publish decision") + run.add_argument("--items", type=Path, help="labelled or unlabelled news items in JSONL") + run.add_argument("--rss", type=Path, action="append", default=[], help="a local RSS or Atom file; repeatable") + run.add_argument( + "--rss-url", action="append", default=[], help="an http(s) feed URL; repeatable; not used by the fixture" + ) + run.add_argument("--x-fixture", type=Path, action="append", default=[], help="X posts in JSONL; repeatable") + run.add_argument("--picks", type=Path, help='JSON object {"ids": [...]} of 3 to 5 kept items') + run.add_argument("--publish", type=Path, help='JSON object {"decisions": {id: publish|hold}}') + run.add_argument("--backend", choices=("auto", "rule", "omni"), default="auto") + run.add_argument( + "--fact-check-worker", + choices=("startlux", "pplx"), + default="startlux", + help="draft worker for fact-check when the backend is omni; pplx needs OMNI_FACTCHECK_BASE_URL", + ) + run.add_argument( + "--allow-unsafe-skip-gates", + action="store_true", + help="UNSAFE. Invent a pick and hold every item so a run can finish without a person. Does not publish.", + ) + run.add_argument("--packet", type=Path, help="write the pre-gate packet here when a human gate stops the run") + latency = commands.add_parser( + "latency-report", + help="summarize Omni request latencies already in a report; optional ceilings fail closed", + ) + latency.add_argument("--report", type=Path, required=True) + latency.add_argument("--p50-max-ms", type=float) + latency.add_argument("--p95-max-ms", type=float) + latency.add_argument("--min-samples", type=int, default=1) + return build + + +def main(argv: list[str] | None = None) -> int: + load_dotenv(ROOT / ".env", override=False) + args = parser().parse_args(argv) + try: + if args.command == "run": + report = _run(args) + else: + report = _latency(args) + except GateRequired as exc: + if args.command == "run" and args.packet is not None and exc.packet is not None: + _write_json(args.packet, exc.packet) + print(exc, file=sys.stderr) + return 2 + except ValueError as exc: + print(exc, file=sys.stderr) + return 2 + except (RuntimeError, OSError, httpx.HTTPError) as exc: + print(exc, file=sys.stderr) + return 1 + print(json.dumps(report, ensure_ascii=False)) + if args.command == "latency-report" and report.get("enforced") and not report.get("passed"): + return 1 + return 0 + + +def _run(args: argparse.Namespace) -> dict[str, Any]: + http = httpx.Client(timeout=15.0, follow_redirects=False, trust_env=False) if args.rss_url else None + try: + items = load_inputs( + items=args.items, + rss_files=args.rss, + rss_urls=args.rss_url, + x_files=args.x_fixture, + http=http, + ) + finally: + if http is not None: + http.close() + backend = _resolve_backend(args.backend, os.environ) + picks = _load_picks(args.picks) if args.picks else None + publish = _load_publish(args.publish) if args.publish else None + interactive = sys.stdin.isatty() and not args.allow_unsafe_skip_gates + pick_hook = _prompt_picks if picks is None and interactive else None + publish_hook = _prompt_publish if publish is None and interactive else None + if backend == "rule": + return run_workflow( + items, + decider=RuleDecider(), + picks=picks, + publish=publish, + pick_hook=pick_hook, + publish_hook=publish_hook, + allow_unsafe_skip_gates=args.allow_unsafe_skip_gates, + workers={"screen": None, "fact_check": None}, + ) + decider = decider_from_env(os.environ, fact_check_worker=args.fact_check_worker) + workers = { + "screen": worker_record( + decider.screen_client.profile, decider.screen_client.base_url, decider.screen_client.model + ), + "fact_check": worker_record( + decider.fact_client.profile, decider.fact_client.base_url, decider.fact_client.model + ), + } + try: + return run_workflow( + items, + decider=decider, + picks=picks, + publish=publish, + pick_hook=pick_hook, + publish_hook=publish_hook, + allow_unsafe_skip_gates=args.allow_unsafe_skip_gates, + workers=workers, + ) + finally: + decider.close() + + +def _resolve_backend(name: str, env: Any) -> str: + if name == "auto": + return "omni" if (env.get("OMNI_BASE_URL") or "").strip() else "rule" + if name == "omni" and not (env.get("OMNI_BASE_URL") or "").strip(): + raise ValueError("backend omni needs OMNI_BASE_URL") + if name == "omni": + require_worker("startlux", fact_check=False) + return name + + +def _load_picks(path: Path) -> list[str]: + data = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(data, dict) or not isinstance(data.get("ids"), list): + raise ValueError("--picks must be a JSON object with an ids array") + return data["ids"] + + +def _load_publish(path: Path) -> dict[str, str]: + data = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(data, dict) or not isinstance(data.get("decisions"), dict): + raise ValueError("--publish must be a JSON object with a decisions object") + return data["decisions"] + + +def _prompt_picks(kept: list[dict[str, Any]]) -> list[str]: + print("Human pick gate. Choose 3 to 5 ids. This gate does not publish.", file=sys.stderr) + for item in kept: + print(f"{item['id']}\t{item['title']}", file=sys.stderr) + try: + raw = input("ids: ") + except EOFError as exc: + raise ValueError("human pick gate closed without a decision") from exc + return [part.strip() for part in raw.split(",") if part.strip()] + + +def _prompt_publish(checked: list[dict[str, Any]]) -> dict[str, str]: + print("Human publish gate. Type publish or hold for each id. Nothing is posted.", file=sys.stderr) + decisions = {} + for row in checked: + print(f"{row['id']} fact-check={row['choice']}", file=sys.stderr) + try: + raw = input("publish or hold: ").strip() + except EOFError as exc: + raise ValueError("human publish gate closed without a decision") from exc + decisions[row["id"]] = raw + return decisions + + +def _latency(args: argparse.Namespace) -> dict[str, Any]: + try: + report = json.loads(args.report.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + raise ValueError(f"--report: {exc}") from exc + if not isinstance(report, dict): + raise ValueError("--report must be a JSON object") + return gate( + omni_call_latencies(report), + p50_max_ms=args.p50_max_ms, + p95_max_ms=args.p95_max_ms, + min_samples=args.min_samples, + ) + + +def _write_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/s1a/content/ingest.py b/s1a/content/ingest.py new file mode 100644 index 0000000..2ac140f --- /dev/null +++ b/s1a/content/ingest.py @@ -0,0 +1,302 @@ +# coding: utf-8 +"""Pluggable ingest. RSS is implemented. Twitter/X is a fixture plus an adapter hook. + +A live X call is not shipped: CI has no X credentials, and this module does not read +``X_BEARER_TOKEN`` or open ``api.x.com``. +""" + +from __future__ import annotations + +import json +import re +from collections.abc import Callable, Iterable +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit +from xml.etree import ElementTree + +import httpx + +SOURCES = ("rss", "x", "fixture") +SCREEN_LABELS = ("keep", "drop") +FACT_LABELS = ("supported", "unsupported", "needs_human") +MAX_RSS_BYTES = 2_000_000 + +LIVE_X = ( + "Twitter/X ingest is not configured. Pass --x-fixture, or pass x_fetch to load_inputs. " + "A live adapter reads X_BEARER_TOKEN itself and sends GET https://api.x.com/2/tweets/search/recent " + "with query and max_results, then maps posts through posts_to_items. " + "s1a.content does not send that request." +) + + +class UnconfiguredXAdapter: + """The default X adapter. ``fetch`` fails closed and does not open a socket.""" + + def fetch(self) -> list[dict[str, Any]]: + raise ValueError(LIVE_X) + + +def normalize_item(row: Any) -> dict[str, Any]: + """Copy the fields the desk may read. Unknown keys, including labels' neighbours, are dropped.""" + if not isinstance(row, dict): + raise ValueError("a content item must be a JSON object") + item: dict[str, Any] = {} + for key in ("id", "title", "summary", "source"): + value = row.get(key) + if not isinstance(value, str) or not value.strip(): + raise ValueError(f"content items need a non-empty string {key}") + item[key] = value.strip() + if item["source"] not in SOURCES: + raise ValueError(f"content source must be one of {', '.join(SOURCES)}") + if "url" in row and row["url"] is not None: + if not isinstance(row["url"], str) or not row["url"].strip(): + raise ValueError("url, when present, must be a non-empty string") + item["url"] = row["url"].strip() + if "label" in row and row["label"] is not None: + if row["label"] not in SCREEN_LABELS: + raise ValueError("screen label must be keep or drop") + item["label"] = row["label"] + if "draft" in row and row["draft"] is not None: + item["draft"] = normalize_draft(row["draft"]) + return item + + +def normalize_draft(row: Any) -> dict[str, Any]: + if not isinstance(row, dict): + raise ValueError("a draft must be a JSON object") + claims = row.get("claims") + sources = row.get("sources") + claims_ok = ( + isinstance(claims, list) and claims and all(isinstance(claim, str) and claim.strip() for claim in claims) + ) + if not claims_ok: + raise ValueError("a draft needs a non-empty list of claim strings") + if not isinstance(sources, list) or not sources: + raise ValueError("a draft needs at least one source") + cleaned_sources = [] + for source in sources: + if not isinstance(source, dict): + raise ValueError("a draft source must be an object with id and text") + sid, text = source.get("id"), source.get("text") + if not isinstance(sid, str) or not sid.strip() or not isinstance(text, str) or not text.strip(): + raise ValueError("a draft source needs non-empty string id and text") + cleaned_sources.append({"id": sid.strip(), "text": text.strip()}) + draft: dict[str, Any] = {"claims": [claim.strip() for claim in claims], "sources": cleaned_sources} + if "label" in row and row["label"] is not None: + if row["label"] not in FACT_LABELS: + raise ValueError("fact-check label must be supported, unsupported or needs_human") + draft["label"] = row["label"] + return draft + + +def public_view(item: dict[str, Any]) -> dict[str, Any]: + """What a decider may see. Screen and fact labels are not copied.""" + view: dict[str, Any] = {key: item[key] for key in ("id", "title", "summary", "source")} + if "url" in item: + view["url"] = item["url"] + draft = item.get("draft") + if draft: + view["draft"] = { + "claims": list(draft["claims"]), + "sources": [{"id": source["id"], "text": source["text"]} for source in draft["sources"]], + } + return view + + +def screen_view(item: dict[str, Any]) -> dict[str, Any]: + view = public_view(item) + view.pop("draft", None) + return view + + +def load_jsonl(path: Path) -> list[dict[str, Any]]: + """News items in JSONL. A bad line names the file and the column, not the line's text.""" + items = [] + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + if not line.strip(): + continue + try: + row = json.loads(line) + except json.JSONDecodeError as exc: + raise ValueError(f"content items {path}: line {line_number}, column {exc.colno}: {exc.msg}") from exc + try: + items.append(normalize_item(row)) + except ValueError as exc: + raise ValueError(f"content items {path}: line {line_number}: {exc}") from exc + return items + + +def posts_to_items(posts: Iterable[dict[str, Any]]) -> list[dict[str, Any]]: + """Map X-shaped posts (``id``, ``text``) or ready news items into desk items.""" + items = [] + for post in posts: + if not isinstance(post, dict): + raise ValueError("an X post must be a JSON object") + if "text" in post and "summary" not in post: + text = post.get("text") + ident = post.get("id") + if not isinstance(ident, str) or not ident.strip() or not isinstance(text, str) or not text.strip(): + raise ValueError("an X post needs non-empty string id and text") + row: dict[str, Any] = { + "id": ident.strip(), + "title": (post.get("title") or text).strip()[:80], + "summary": text.strip(), + "source": "x", + } + if post.get("url"): + row["url"] = post["url"] + if post.get("label"): + row["label"] = post["label"] + items.append(normalize_item(row)) + else: + items.append(normalize_item(post)) + return items + + +def load_x_fixture(path: Path) -> list[dict[str, Any]]: + posts = [] + for line_number, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): + if not line.strip(): + continue + try: + posts.append(json.loads(line)) + except json.JSONDecodeError as exc: + raise ValueError(f"x fixture {path}: line {line_number}, column {exc.colno}: {exc.msg}") from exc + try: + return posts_to_items(posts) + except ValueError as exc: + raise ValueError(f"x fixture {path}: {exc}") from exc + + +def _local(tag: str) -> str: + return tag.rsplit("}", 1)[-1] + + +def _text(element: ElementTree.Element | None) -> str: + if element is None: + return "" + raw = "".join(element.itertext()) + raw = re.sub(r"<[^>]+>", " ", raw) + return re.sub(r"\s+", " ", raw).strip() + + +def _child(parent: ElementTree.Element, name: str) -> ElementTree.Element | None: + for child in list(parent): + if _local(child.tag) == name: + return child + return None + + +def parse_feed(xml_text: str, *, origin: str) -> list[dict[str, Any]]: + """RSS 2.0 items or Atom entries. Ids prefer guid, then id, then link.""" + try: + root = ElementTree.fromstring(xml_text) + except ElementTree.ParseError as exc: + raise ValueError(f"feed {origin}: {exc}") from exc + parents = [element for element in root.iter() if _local(element.tag) in {"channel", "feed"}] + if not parents: + parents = [root] + rows = [] + for parent in parents: + for element in list(parent): + kind = _local(element.tag) + if kind not in {"item", "entry"}: + continue + title = _text(_child(element, "title")) + summary = ( + _text(_child(element, "description")) + or _text(_child(element, "summary")) + or _text(_child(element, "content")) + ) + link = _link(element) + guid = _text(_child(element, "guid")) or _text(_child(element, "id")) or link + if not title or not summary or not guid: + raise ValueError(f"feed {origin}: an entry needs a title, a summary and an id or link") + row: dict[str, Any] = {"id": guid, "title": title, "summary": summary, "source": "rss"} + if link: + row["url"] = link + rows.append(normalize_item(row)) + if not rows: + raise ValueError(f"feed {origin}: no entries") + return rows + + +def _link(element: ElementTree.Element) -> str: + for child in list(element): + if _local(child.tag) != "link": + continue + href = (child.get("href") or "").strip() + if href: + return href + if child.text and child.text.strip(): + return child.text.strip() + return "" + + +def load_rss_file(path: Path) -> list[dict[str, Any]]: + return parse_feed(path.read_text(encoding="utf-8"), origin=str(path)) + + +def check_http_url(url: str) -> str: + parsed = urlsplit(url.strip()) + if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.username or parsed.password: + raise ValueError("a feed URL must be http(s) without credentials") + return url.strip() + + +def load_rss_url(url: str, *, client: httpx.Client) -> list[dict[str, Any]]: + """GET one feed. The caller owns the client; tests pass a mock transport.""" + target = check_http_url(url) + try: + response = client.get(target) + except httpx.HTTPError as exc: + raise RuntimeError(f"feed {target} was not fetched: {type(exc).__name__}") from exc + if response.status_code != 200: + raise RuntimeError(f"feed {target} returned HTTP {response.status_code}") + body = response.content + if len(body) > MAX_RSS_BYTES: + raise ValueError(f"feed {target} is larger than {MAX_RSS_BYTES} bytes") + try: + text = body.decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"feed {target} is not utf-8") from exc + return parse_feed(text, origin=target) + + +def ensure_unique(items: list[dict[str, Any]]) -> None: + seen: set[str] = set() + for item in items: + if item["id"] in seen: + raise ValueError(f"duplicate content id {item['id']!r}") + seen.add(item["id"]) + + +def load_inputs( + *, + items: Path | None = None, + rss_files: list[Path] | None = None, + rss_urls: list[str] | None = None, + x_files: list[Path] | None = None, + x_fetch: Callable[[], list[dict[str, Any]]] | None = None, + http: httpx.Client | None = None, +) -> list[dict[str, Any]]: + """Fixture JSONL, then RSS files, then RSS URLs, then X fixtures, then an optional adapter.""" + loaded: list[dict[str, Any]] = [] + if items is not None: + loaded.extend(load_jsonl(items)) + for path in rss_files or []: + loaded.extend(load_rss_file(path)) + if rss_urls: + if http is None: + raise ValueError("rss urls need an HTTP client") + for url in rss_urls: + loaded.extend(load_rss_url(url, client=http)) + for path in x_files or []: + loaded.extend(load_x_fixture(path)) + if x_fetch is not None: + loaded.extend(posts_to_items(x_fetch())) + if not loaded: + raise ValueError("content workflow needs at least one item") + ensure_unique(loaded) + return loaded diff --git a/s1a/content/latency.py b/s1a/content/latency.py new file mode 100644 index 0000000..a04cf33 --- /dev/null +++ b/s1a/content/latency.py @@ -0,0 +1,92 @@ +# coding: utf-8 +"""Latency summary and the hook for a later Public231-C1-style ceiling. + +Nothing here contacts a worker. A workflow run records per-request wall time and leaves the +ceiling unattached. ``gate`` is what a later operator command calls once measured samples exist. +""" + +from __future__ import annotations + +import math +from typing import Any + +CLOCK = "client_request_wall_ms" +PROCEDURE = ( + "Public231 C1 later gate: drop a warmup of 5, keep concurrency at 1, repeat three measured rounds, " + "then compare p50 and p95 of successful request latencies. " + "This client's clock is client_request_wall_ms, from the start of the POST through the response body. " + "System1-Omni's bench clock is post_to_body_json_and_schema_validation; do not treat the two as the same number. " + "No ceiling is pinned for the content fixture. A p50 from another workload is not this fixture's budget." +) + + +def percentile(samples: list[float], percent: float) -> float: + """Nearest-rank percentile. ``percent`` is in 0–100. Rank is ``ceil(percent/100 * n)``, clamped to the sample.""" + if not samples: + raise ValueError("percentile needs at least one sample") + if not 0 < percent <= 100: + raise ValueError("percent must be in (0, 100]") + ordered = sorted(samples) + rank = max(1, math.ceil(percent / 100 * len(ordered))) + return ordered[rank - 1] + + +def summarize_ms(samples: list[float]) -> dict[str, Any]: + if not samples: + return {"n": 0, "min_ms": None, "p50_ms": None, "p95_ms": None, "max_ms": None, "clock": CLOCK} + return { + "n": len(samples), + "min_ms": min(samples), + "p50_ms": percentile(samples, 50), + "p95_ms": percentile(samples, 95), + "max_ms": max(samples), + "clock": CLOCK, + } + + +def gate( + samples: list[float], + *, + p50_max_ms: float | None = None, + p95_max_ms: float | None = None, + min_samples: int = 1, +) -> dict[str, Any]: + """Enforce optional ceilings. With neither ceiling set, the gate is not attached and does not pass or fail. + + When a ceiling is set, fewer than ``min_samples`` fails closed. Zero-millisecond rule-baseline rows must + not be passed in: they would clear a ceiling without a worker. + """ + summary = summarize_ms(samples) + enforced = p50_max_ms is not None or p95_max_ms is not None + if not enforced: + return {"enforced": False, "passed": None, "summary": summary, "failures": [], "procedure": PROCEDURE} + failures: list[str] = [] + if min_samples < 1: + raise ValueError("min_samples must be at least 1 when a ceiling is set") + if len(samples) < min_samples: + failures.append(f"need at least {min_samples} omni request samples, got {len(samples)}") + else: + if p50_max_ms is not None and summary["p50_ms"] > p50_max_ms: + failures.append(f"p50 {summary['p50_ms']} ms is above {p50_max_ms} ms") + if p95_max_ms is not None and summary["p95_ms"] > p95_max_ms: + failures.append(f"p95 {summary['p95_ms']} ms is above {p95_max_ms} ms") + return { + "enforced": True, + "passed": not failures, + "summary": summary, + "failures": failures, + "procedure": PROCEDURE, + } + + +def omni_call_latencies(report: dict[str, Any]) -> list[float]: + """Successful Omni request latencies from a workflow report. Rule rows are left out.""" + samples: list[float] = [] + for stage in ("screen", "fact_check"): + block = report.get(stage) or {} + for record in block.get("records") or []: + if record.get("backend") != "omni": + continue + for call in record.get("calls") or []: + samples.append(float(call["latency_ms"])) + return samples diff --git a/s1a/content/omni_client.py b/s1a/content/omni_client.py new file mode 100644 index 0000000..f3eabfe --- /dev/null +++ b/s1a/content/omni_client.py @@ -0,0 +1,366 @@ +# coding: utf-8 +"""HTTP client for draft System1-Omni workers that speak ``POST /v1/systemone``. + +The default worker is StartLux Instant 4B from InferMatrixLab/system1-omni draft pull request 164. +Fact-check may switch to the PPLX draft worker from pull request 166, on its own base URL. +Clef, pull request 165, is refused on this path. None of those pull requests need to be merged: +the client only needs a base URL that already serves the contract. + +Each question is its own short prefill, Public231-style: one ``state`` string and one question. +StartLux is sent ``think_mode: never``. PPLX is sent ``thinking: {enabled: false}``. +Those are different draft contracts; one body is not sent to both. +""" + +from __future__ import annotations + +import math +import time +import uuid +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from typing import Any +from urllib.parse import urlsplit + +import httpx + +from s1a.content.rules import FACT_SCORE, SCREEN_SCORE, fact_questions, fact_state, screen_questions, screen_state + +CLEF_PR = 165 +DEFAULT_TIMEOUT_S = 30.0 + + +@dataclass(frozen=True) +class WorkerProfile: + key: str + model: str + pr: int + hot_path: bool + thinking: str + role: str + + @property + def pr_url(self) -> str: + return f"https://github.com/InferMatrixLab/system1-omni/pull/{self.pr}" + + +WORKERS = { + "startlux": WorkerProfile( + key="startlux", + model="StartLux-Decision-v1.1-4B-Instant", + pr=164, + hot_path=True, + thinking="think_mode", + role="default screen and fact-check", + ), + "pplx": WorkerProfile( + key="pplx", + model="perplexity-ai/pplx-decider-v1.1-27b", + pr=166, + hot_path=False, + thinking="thinking_object", + role="optional heavier fact-check", + ), +} + + +def require_worker(name: str, *, fact_check: bool) -> WorkerProfile: + if name == "clef": + raise ValueError( + "Clef is not on the content hot path " + f"(InferMatrixLab/system1-omni pull request {CLEF_PR}). " + "Use startlux for screen and fact-check, or pplx for fact-check only." + ) + profile = WORKERS.get(name) + if profile is None: + raise ValueError(f"unknown worker {name!r}; choose startlux or pplx") + if not fact_check and name != "startlux": + raise ValueError("the screen step uses the StartLux Instant 4B draft worker; pplx is fact-check only") + return profile + + +def check_base_url(url: str) -> str: + parsed = urlsplit(url.strip()) + if ( + parsed.scheme not in {"http", "https"} + or not parsed.netloc + or parsed.username + or parsed.password + or parsed.query + or parsed.fragment + or parsed.path not in {"", "/"} + ): + raise ValueError("an Omni base URL must be http(s) without credentials, a query, or a path") + return url.strip().rstrip("/") + + +def timeout_from_env(env: Mapping[str, str]) -> float: + raw = (env.get("OMNI_TIMEOUT_S") or str(DEFAULT_TIMEOUT_S)).strip() + try: + value = float(raw) + except ValueError as exc: + raise ValueError("OMNI_TIMEOUT_S must be a finite number above 0") from exc + if not math.isfinite(value) or value <= 0: + raise ValueError("OMNI_TIMEOUT_S must be a finite number above 0") + return value + + +def worker_record(profile: WorkerProfile, base_url: str, model: str) -> dict[str, Any]: + return { + "worker": profile.key, + "model": model, + "pr": profile.pr, + "pr_url": profile.pr_url, + "draft": True, + "merged_required": False, + "base_url": base_url, + "hot_path": profile.hot_path, + "prefill": "short", + "questions": ["choice", "score"], + "thinking": profile.thinking, + } + + +class OmniClient: + """One draft worker. Synchronous, one POST per question, no retry. + + Retries would blur the latency a later C1 gate reads. A failed status raises and does not + include the response body, which can echo the submitted state. + """ + + def __init__( + self, + base_url: str, + *, + profile: WorkerProfile, + model: str | None = None, + api_key: str | None = None, + timeout_s: float = DEFAULT_TIMEOUT_S, + transport: httpx.BaseTransport | None = None, + ) -> None: + if not math.isfinite(timeout_s) or timeout_s <= 0: + raise ValueError("timeout_s must be a finite number above 0") + self.base_url = check_base_url(base_url) + self.profile = profile + self.model = model or profile.model + self._timeout_s = timeout_s + headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} + self._client = httpx.Client( + timeout=timeout_s, + headers=headers, + transport=transport, + follow_redirects=False, + trust_env=False, + ) + + def close(self) -> None: + self._client.close() + + def health(self) -> dict[str, Any]: + try: + response = self._client.get(f"{self.base_url}/health") + except httpx.HTTPError as exc: + raise RuntimeError(f"no Omni worker at {self.base_url}: {type(exc).__name__}") from exc + if response.status_code != 200: + raise RuntimeError(f"Omni health at {self.base_url} returned HTTP {response.status_code}") + try: + body = response.json() + except ValueError as exc: + raise RuntimeError(f"Omni health at {self.base_url} was not JSON") from exc + if not isinstance(body, dict): + raise RuntimeError(f"Omni health at {self.base_url} was not a JSON object") + return body + + def ask(self, state: str, question_id: str, question: dict[str, Any]) -> tuple[dict[str, Any], int, str]: + if not state.strip(): + raise ValueError("an Omni prefill state must not be empty") + if len(question) == 0: + raise ValueError("an Omni question must not be empty") + body = self._body(state, question_id, question) + request_id = str(uuid.uuid4()) + started = time.perf_counter() + try: + response = self._client.post( + f"{self.base_url}/v1/systemone", + json=body, + headers={"X-Request-Id": request_id}, + ) + except httpx.HTTPError as exc: + raise RuntimeError(f"Omni request to {self.base_url} failed: {type(exc).__name__}") from exc + latency_ms = round((time.perf_counter() - started) * 1000) + if response.status_code != 200: + raise RuntimeError(f"Omni {self.base_url} returned HTTP {response.status_code} for question {question_id}") + try: + payload = response.json() + except ValueError as exc: + raise RuntimeError(f"Omni {self.base_url} returned a non-JSON body for {question_id}") from exc + answers = payload.get("answers") if isinstance(payload, dict) else None + answer = answers.get(question_id) if isinstance(answers, dict) else None + if not isinstance(answer, dict): + raise RuntimeError(f"Omni {self.base_url} returned no answer for {question_id}") + if isinstance(payload, dict) and isinstance(payload.get("model"), str): + answer = dict(answer) + answer["_served_model"] = payload["model"] + return answer, latency_ms, request_id + + def _body(self, state: str, question_id: str, question: dict[str, Any]) -> dict[str, Any]: + body: dict[str, Any] = { + "model": self.model, + "state": state, + "questions": {question_id: question}, + } + if self.profile.thinking == "think_mode": + body["think_mode"] = "never" + elif self.profile.thinking == "thinking_object": + body["thinking"] = {"enabled": False} + else: + raise ValueError(f"worker {self.profile.key} has no thinking contract") + return body + + +def _finite_number(value: Any, *, what: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(float(value)): + raise RuntimeError(f"Omni {what} was not a finite number") + return float(value) + + +def parse_choice(answer: dict[str, Any], criteria: dict[str, str]) -> tuple[str, dict[str, float]]: + choice = answer.get("choice") + if not isinstance(choice, str) or choice not in criteria: + raise RuntimeError("Omni choice was outside the offered keys") + raw = answer.get("probabilities") + if not isinstance(raw, dict): + raise RuntimeError("Omni choice was missing probabilities") + probabilities = {key: _finite_number(raw.get(key), what=f"probability {key}") for key in criteria} + return choice, probabilities + + +def parse_score(answer: dict[str, Any], rubric: list[str]) -> tuple[float, int, dict[str, float], str]: + score = _finite_number(answer.get("score"), what="score") + if score < 0: + raise RuntimeError("Omni score was negative") + raw = answer.get("probabilities") + probabilities: dict[str, float] = {} + if raw is not None: + if not isinstance(raw, dict): + raise RuntimeError("Omni score probabilities were not an object") + for index in range(len(rubric)): + key = str(index) + if key in raw: + probabilities[key] = _finite_number(raw[key], what=f"score probability {key}") + if probabilities: + level = max(range(len(rubric)), key=lambda index: probabilities.get(str(index), float("-inf"))) + else: + level = min(len(rubric) - 1, max(0, int(round(score)))) + return score, level, probabilities, rubric[level] + + +class OmniDecider: + """Screen on one client, fact-check on the same client or a heavier draft worker.""" + + name = "omni" + + def __init__(self, screen: OmniClient, fact: OmniClient) -> None: + self.screen_client = screen + self.fact_client = fact + + def close(self) -> None: + self.screen_client.close() + if self.fact_client is not self.screen_client: + self.fact_client.close() + + def screen(self, item: dict[str, Any]) -> dict[str, Any]: + return _pair(self.screen_client, screen_state(item), screen_questions(), "screen", "relevance", SCREEN_SCORE) + + def fact_check(self, item: dict[str, Any]) -> dict[str, Any]: + return _pair(self.fact_client, fact_state(item), fact_questions(), "verdict", "support", FACT_SCORE) + + +def _pair( + client: OmniClient, + state: str, + questions: dict[str, dict[str, Any]], + choice_id: str, + score_id: str, + rubric: list[str], +) -> dict[str, Any]: + """Ask the choice and the score as two short prefills. Each is judged against the label later.""" + choice_answer, choice_ms, choice_request = client.ask(state, choice_id, questions[choice_id]) + score_answer, score_ms, score_request = client.ask(state, score_id, questions[score_id]) + criteria = questions[choice_id]["criteria"] + choice, probabilities = parse_choice(choice_answer, criteria) + score, level, score_probabilities, score_name = parse_score(score_answer, rubric) + served = choice_answer.get("_served_model") + model = served if isinstance(served, str) else client.model + return { + "choice": choice, + "probabilities": probabilities, + "score": score, + "score_level": level, + "score_name": score_name, + "score_probabilities": score_probabilities, + "latency_ms": choice_ms + score_ms, + "backend": "omni", + "model": model, + "requested_model": client.model, + "worker": client.profile.key, + "calls": [ + {"question": choice_id, "latency_ms": choice_ms, "request_id": choice_request}, + {"question": score_id, "latency_ms": score_ms, "request_id": score_request}, + ], + } + + +def decider_from_env( + env: Mapping[str, str], + *, + fact_check_worker: str = "startlux", + handler: Callable[[httpx.Request], httpx.Response] | None = None, +) -> OmniDecider: + """Build the screen client from ``OMNI_BASE_URL`` and, for PPLX, a second URL. + + ``handler`` wraps a fresh mock transport per client so tests never open a socket. + """ + screen_profile = require_worker("startlux", fact_check=False) + fact_profile = require_worker(fact_check_worker, fact_check=True) + base_url = check_base_url(env.get("OMNI_BASE_URL") or "") + timeout_s = timeout_from_env(env) + api_key = (env.get("OMNI_API_KEY") or "").strip() or None + screen_model = (env.get("OMNI_MODEL") or "").strip() or screen_profile.model + fact_url = (env.get("OMNI_FACTCHECK_BASE_URL") or "").strip() + fact_model = (env.get("OMNI_FACTCHECK_MODEL") or "").strip() or fact_profile.model + if fact_profile.key == "pplx" and not fact_url: + raise ValueError( + "PPLX fact-check needs OMNI_FACTCHECK_BASE_URL. " + "Draft pull request 166 is a different worker from StartLux and is not assumed to share its port." + ) + if fact_profile.key == "startlux" and not fact_url: + fact_url = base_url + fact_model = (env.get("OMNI_FACTCHECK_MODEL") or "").strip() or screen_model + else: + fact_url = check_base_url(fact_url) + + def transport() -> httpx.BaseTransport | None: + if handler is None: + return None + return httpx.MockTransport(handler) + + screen = OmniClient( + base_url, + profile=screen_profile, + model=screen_model, + api_key=api_key, + timeout_s=timeout_s, + transport=transport(), + ) + if fact_url == screen.base_url and fact_profile.key == screen_profile.key and fact_model == screen.model: + return OmniDecider(screen, screen) + fact_key = (env.get("OMNI_FACTCHECK_API_KEY") or env.get("OMNI_API_KEY") or "").strip() or None + fact = OmniClient( + fact_url, + profile=fact_profile, + model=fact_model, + api_key=fact_key, + timeout_s=timeout_s, + transport=transport(), + ) + return OmniDecider(screen, fact) diff --git a/s1a/content/rules.py b/s1a/content/rules.py new file mode 100644 index 0000000..04ab979 --- /dev/null +++ b/s1a/content/rules.py @@ -0,0 +1,176 @@ +# coding: utf-8 +"""Screen and fact-check questions, plus the offline lexical stand-in. + +The stand-in is the same kind of baseline as ``ticket_router``'s keyword rule: it makes the +fixture path deterministic without a model. It is not a quality claim for Jev or for a draft +System1-Omni worker. Labels on the shipped sample match this stand-in on purpose. +""" + +from __future__ import annotations + +import re +from typing import Any + +# Whole words only, so "supermodel", "chipotle" and "agency" are not hotspots. +KEEP = re.compile(r"\b(?:models?|agents?|inference|chips?|research|checkpoints?|llms?)\b", re.IGNORECASE) +_WORD = re.compile(r"[a-z0-9]{5,}") +_NUMBER = re.compile(r"\d+(?:\.\d+)?") + +STATE_LIMIT = 1500 +SOURCE_LIMIT = 400 + +SCREEN_CHOICES = { + "keep": "Relevant AI, model, agent, inference, chip, checkpoint or research hotspot", + "drop": "Not a relevant hotspot for this desk", +} +SCREEN_SCORE = ["Not relevant", "Relevant hotspot"] +FACT_CHOICES = { + "supported": "The sources state the claim", + "unsupported": "A name or number in the claim is absent from the sources", + "needs_human": "The sources do not settle the claim", +} +FACT_SCORE = ["Unsupported", "Needs a human", "Supported"] +FACT_LEVEL = {"unsupported": 0, "needs_human": 1, "supported": 2} + +SCREEN_INSTRUCTIONS = ( + "Keep a relevant hotspot for an AI-agent news desk: models, agents, inference, chips, " + "checkpoints or research. Drop sports, weather, recipes, traffic, celebrity and anything without that hook." +) +SCREEN_SCORE_INSTRUCTIONS = "How relevant is this item to an AI-agent news desk?" +FACT_INSTRUCTIONS = ( + "Judge the claims against the sources only. supported: the sources state the claim. " + "unsupported: a name or number in the claim is absent from the sources. " + "needs_human: the sources do not settle the claim." +) +FACT_SCORE_INSTRUCTIONS = "How well do the sources support the claims?" + + +def lexical_screen(item: dict[str, Any]) -> str: + """Keep when the title or summary names a desk topic; otherwise drop.""" + text = f"{item.get('title', '')}\n{item.get('summary', '')}" + return "keep" if KEEP.search(text) else "drop" + + +def lexical_fact(draft: dict[str, Any]) -> str: + """Supported when each claim's longer words sit in the sources and every number does too. + + A missing number is unsupported even when the wording overlaps. Thin overlap is needs_human. + An empty claim list or an empty source list is needs_human. This reads no label. + """ + claims = draft.get("claims") or [] + sources = draft.get("sources") or [] + if not claims or not sources: + return "needs_human" + blob = "\n".join(str(source.get("text", "")) for source in sources).casefold() + unsupported = False + overlapped = 0 + for claim in claims: + folded = str(claim).casefold() + numbers = _NUMBER.findall(folded) + if any(number not in blob for number in numbers): + unsupported = True + words = _WORD.findall(folded) + if words and sum(word in blob for word in words) / len(words) >= 0.6: + overlapped += 1 + if unsupported: + return "unsupported" + if overlapped == len(claims): + return "supported" + return "needs_human" + + +def screen_level(choice: str) -> int: + if choice not in SCREEN_CHOICES: + raise ValueError(f"screen choice must be keep or drop, got {choice!r}") + return 1 if choice == "keep" else 0 + + +def fact_level(choice: str) -> int: + if choice not in FACT_LEVEL: + raise ValueError(f"fact-check choice must be one of {', '.join(FACT_LEVEL)}, got {choice!r}") + return FACT_LEVEL[choice] + + +def one_hot(keys: list[str], chosen: str) -> dict[str, float]: + if chosen not in keys: + raise ValueError(f"{chosen!r} is not one of {keys}") + return {key: (1.0 if key == chosen else 0.0) for key in keys} + + +def choice_question(instructions: str, criteria: dict[str, str]) -> dict[str, Any]: + return {"type": "choice", "instructions": instructions, "criteria": dict(criteria)} + + +def score_question(instructions: str, criteria: list[str]) -> dict[str, Any]: + return {"type": "score", "instructions": instructions, "criteria": list(criteria)} + + +def screen_questions() -> dict[str, dict[str, Any]]: + """One question per request. Callers send ``screen`` and ``relevance`` as two prefills.""" + return { + "screen": choice_question(SCREEN_INSTRUCTIONS, SCREEN_CHOICES), + "relevance": score_question(SCREEN_SCORE_INSTRUCTIONS, SCREEN_SCORE), + } + + +def fact_questions() -> dict[str, dict[str, Any]]: + return { + "verdict": choice_question(FACT_INSTRUCTIONS, FACT_CHOICES), + "support": score_question(FACT_SCORE_INSTRUCTIONS, FACT_SCORE), + } + + +def _clip(text: str) -> str: + text = text.strip() + if len(text) <= STATE_LIMIT: + return text + return text[: STATE_LIMIT - 1].rstrip() + "…" + + +def screen_state(item: dict[str, Any]) -> str: + """Short prefill. Labels, drafts and urls stay out of the text the worker reads.""" + return _clip(f"Title: {item['title'].strip()}\nSummary: {item['summary'].strip()}\nSource: {item['source']}") + + +def fact_state(item: dict[str, Any]) -> str: + """Short prefill of the human's claims and sources. No label line.""" + draft = item["draft"] + lines = [f"Title: {item['title'].strip()}", "Claims:"] + lines.extend(f"- {str(claim).strip()}" for claim in draft["claims"]) + lines.append("Sources:") + for source in draft["sources"]: + lines.append(f"- [{source['id']}] {str(source['text']).strip()[:SOURCE_LIMIT]}") + return _clip("\n".join(lines)) + + +class RuleDecider: + """Offline keep/drop and fact verdicts. No network and no label reads.""" + + name = "rule" + + def screen(self, item: dict[str, Any]) -> dict[str, Any]: + choice = lexical_screen(item) + level = screen_level(choice) + return _ruled(choice, list(SCREEN_CHOICES), level, SCREEN_SCORE) + + def fact_check(self, item: dict[str, Any]) -> dict[str, Any]: + choice = lexical_fact(item["draft"]) + level = fact_level(choice) + return _ruled(choice, list(FACT_CHOICES), level, FACT_SCORE) + + +def _ruled(choice: str, keys: list[str], level: int, rubric: list[str]) -> dict[str, Any]: + return { + "choice": choice, + "probabilities": one_hot(keys, choice), + "score": float(level), + "score_level": level, + "score_name": rubric[level], + "score_probabilities": one_hot([str(i) for i in range(len(rubric))], str(level)), + "latency_ms": 0, + "backend": "rule", + "model": None, + "requested_model": None, + "worker": None, + "calls": [], + } diff --git a/s1a/content/workflow.py b/s1a/content/workflow.py new file mode 100644 index 0000000..25301be --- /dev/null +++ b/s1a/content/workflow.py @@ -0,0 +1,351 @@ +# coding: utf-8 +"""Five-stage desk run. Human pick and human publish are gates. Nothing is posted. + +Stages, in order: ingest (already a list of items), screen, human pick, fact-check, human publish. +A missing gate raises ``GateRequired`` unless ``allow_unsafe_skip_gates`` is set. That flag is unsafe: +it invents a pick and holds every item. It still does not publish. +""" + +from __future__ import annotations + +import hashlib +import json +from collections import Counter +from collections.abc import Callable, Mapping, Sequence +from typing import Any, Protocol + +from s1a.content.ingest import normalize_item, public_view, screen_view +from s1a.content.latency import CLOCK, PROCEDURE +from s1a.content.rules import FACT_LEVEL, SCREEN_CHOICES, fact_level, screen_level + +STAGES = ("ingest", "screen", "human_pick", "fact_check", "human_publish") +PUBLISH_DECISIONS = ("publish", "hold") + + +class Decider(Protocol): + name: str + + def screen(self, item: dict[str, Any]) -> dict[str, Any]: ... + + def fact_check(self, item: dict[str, Any]) -> dict[str, Any]: ... + + +class GateRequired(ValueError): + """A human gate has no decision. ``packet`` is the work finished before the stop, if any.""" + + def __init__(self, message: str, packet: dict[str, Any] | None = None) -> None: + super().__init__(message) + self.packet = packet + + +def items_sha256(items: Sequence[Mapping[str, Any]]) -> str: + payload = json.dumps(list(items), ensure_ascii=False, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +def run_workflow( + items: Sequence[Mapping[str, Any]], + *, + decider: Decider, + picks: Sequence[str] | None = None, + publish: Mapping[str, str] | None = None, + pick_hook: Callable[[list[dict[str, Any]]], list[str]] | None = None, + publish_hook: Callable[[list[dict[str, Any]]], Mapping[str, str]] | None = None, + allow_unsafe_skip_gates: bool = False, + workers: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + """Walk the five stages. ``items`` are already ingested. Labels are read only when scoring. + + Explicit ``picks`` / ``publish`` win over hooks. Hooks win over the unsafe skip. The unsafe skip + never wins over a decision the caller actually supplied. + """ + if not items: + raise ValueError("content workflow needs at least one item") + rows = [normalize_item(dict(item)) for item in items] + seen: set[str] = set() + for item in rows: + if item["id"] in seen: + raise ValueError(f"duplicate content id {item['id']!r}") + seen.add(item["id"]) + + screen_records = [_score_screen(item, decider.screen(screen_view(item))) for item in rows] + screen = _stage_summary(screen_records, kept_key="choice") + kept = [record for record in screen_records if record["choice"] == "keep"] + packet_base = _packet_base(rows, decider.name, workers) + + pick_ids, pick_source, pick_unsafe = _resolve_picks( + kept, picks=picks, pick_hook=pick_hook, allow_unsafe_skip_gates=allow_unsafe_skip_gates + ) + if pick_ids is None: + packet = { + **packet_base, + "stopped_at": "human_pick", + "screen": screen, + "human_pick": None, + "fact_check": None, + "human_publish": _unpublished(), + } + raise GateRequired( + "human pick gate: pass picks, answer the prompt, or pass --allow-unsafe-skip-gates " + "(unsafe: it invents a pick and holds every item; it still does not publish)", + packet, + ) + + by_id = {item["id"]: item for item in rows} + fact_records = [] + for ident in pick_ids: + item = by_id[ident] + if not item.get("draft"): + fact_records.append(_no_draft(ident)) + continue + fact_records.append(_score_fact(item, decider.fact_check(public_view(item)))) + fact = _stage_summary(fact_records, kept_key=None) + + decisions, publish_source, publish_unsafe = _resolve_publish( + fact_records, + publish=publish, + publish_hook=publish_hook, + allow_unsafe_skip_gates=allow_unsafe_skip_gates, + ) + if decisions is None: + preview = ", ".join(f"{record['id']}={record['choice']}" for record in fact_records) + packet = { + **packet_base, + "stopped_at": "human_publish", + "screen": screen, + "human_pick": {"ids": pick_ids, "source": pick_source, "unsafe": pick_unsafe}, + "fact_check": fact, + "human_publish": _unpublished(), + } + raise GateRequired( + "human publish gate: pass --publish for " + f"{', '.join(pick_ids)} (fact-check {preview}). Nothing was published", + packet, + ) + + return { + "schema": "content-workflow-v1", + "stages": list(STAGES), + "backend": decider.name, + "unsafe_skip": pick_unsafe or publish_unsafe, + "auto_publish": False, + "ingest": packet_base["ingest"], + "screen": screen, + "human_pick": {"ids": pick_ids, "source": pick_source, "unsafe": pick_unsafe}, + "fact_check": fact, + "human_publish": { + "decisions": decisions, + "source": publish_source, + "unsafe": publish_unsafe, + "executed": [], + "auto_publish": False, + "note": "Recorded the human decision. Nothing was published.", + }, + "workers": packet_base["workers"], + "latency_gate": {"attached": False, "clock": CLOCK, "procedure": PROCEDURE}, + } + + +def _packet_base(rows: list[dict[str, Any]], backend: str, workers: Mapping[str, Any] | None) -> dict[str, Any]: + counts = Counter(item["source"] for item in rows) + return { + "schema": "content-workflow-packet-v1", + "backend": backend, + "auto_publish": False, + "ingest": { + "count": len(rows), + "sources": {name: counts.get(name, 0) for name in ("rss", "x", "fixture")}, + "sha256": items_sha256(rows), + }, + "workers": dict(workers) if workers else {"screen": None, "fact_check": None}, + "latency_gate": {"attached": False, "clock": CLOCK, "procedure": PROCEDURE}, + } + + +def _unpublished() -> dict[str, Any]: + return { + "decisions": {}, + "source": None, + "unsafe": False, + "executed": [], + "auto_publish": False, + "note": "Recorded no publish decision. Nothing was published.", + } + + +def _resolve_picks( + kept: list[dict[str, Any]], + *, + picks: Sequence[str] | None, + pick_hook: Callable[[list[dict[str, Any]]], list[str]] | None, + allow_unsafe_skip_gates: bool, +) -> tuple[list[str] | None, str, bool]: + kept_ids = [record["id"] for record in kept] + if picks is not None: + return _validate_picks(list(picks), kept_ids), "explicit", False + if pick_hook is not None: + shown = [ + {"id": record["id"], "title": record["title"], "summary": record["summary"], "source": record["source"]} + for record in kept + ] + return _validate_picks(list(pick_hook(shown)), kept_ids), "hook", False + if allow_unsafe_skip_gates: + if not kept_ids: + raise ValueError("screen kept nothing; the unsafe skip has no item to invent") + invented = kept_ids if len(kept_ids) < 3 else kept_ids[:3] + return _validate_picks(invented, kept_ids), "unsafe-skip", True + return None, "", False + + +def _validate_picks(picks: list[str], kept_ids: list[str]) -> list[str]: + if not picks or any(not isinstance(ident, str) or not ident.strip() for ident in picks): + raise ValueError("a human pick is a list of non-empty ids") + cleaned = [ident.strip() for ident in picks] + if len(cleaned) != len(set(cleaned)): + raise ValueError("a human pick repeats an id") + missing = [ident for ident in cleaned if ident not in kept_ids] + if missing: + raise ValueError(f"human pick {missing[0]!r} was not kept by the screen") + if len(kept_ids) >= 3 and not 3 <= len(cleaned) <= 5: + raise ValueError("the human pick is 3 to 5 kept items") + if len(kept_ids) < 3 and cleaned != kept_ids: + raise ValueError("fewer than 3 items were kept; the human pick must be every kept id") + return cleaned + + +def _resolve_publish( + fact_records: list[dict[str, Any]], + *, + publish: Mapping[str, str] | None, + publish_hook: Callable[[list[dict[str, Any]]], Mapping[str, str]] | None, + allow_unsafe_skip_gates: bool, +) -> tuple[dict[str, str] | None, str, bool]: + expected = [record["id"] for record in fact_records] + if publish is not None: + return _validate_publish(dict(publish), expected), "explicit", False + if publish_hook is not None: + shown = [ + {"id": record["id"], "choice": record["choice"], "score_name": record["score_name"]} + for record in fact_records + ] + return _validate_publish(dict(publish_hook(shown)), expected), "hook", False + if allow_unsafe_skip_gates: + return {ident: "hold" for ident in expected}, "unsafe-skip", True + return None, "", False + + +def _validate_publish(decisions: Mapping[str, str], expected: list[str]) -> dict[str, str]: + if set(decisions) != set(expected): + raise ValueError("the publish decision must name each picked id once, and no other id") + cleaned: dict[str, str] = {} + for ident in expected: + value = decisions[ident] + if value not in PUBLISH_DECISIONS: + raise ValueError("a publish decision is publish or hold") + cleaned[ident] = value + return cleaned + + +def _score_screen(item: Mapping[str, Any], decision: Mapping[str, Any]) -> dict[str, Any]: + expected = item.get("label") + expected_level = screen_level(expected) if expected in SCREEN_CHOICES else None + record = _attach( + item, decision, expected_choice=expected if expected in SCREEN_CHOICES else None, expected_level=expected_level + ) + record["title"] = item["title"] + record["summary"] = item["summary"] + record["source"] = item["source"] + return record + + +def _score_fact(item: Mapping[str, Any], decision: Mapping[str, Any]) -> dict[str, Any]: + draft = item.get("draft") or {} + expected = draft.get("label") + expected_level = fact_level(expected) if expected in FACT_LEVEL else None + return _attach( + item, + decision, + expected_choice=expected if expected in FACT_LEVEL else None, + expected_level=expected_level, + ) + + +def _no_draft(ident: str) -> dict[str, Any]: + return { + "id": ident, + "choice": "needs_human", + "probabilities": {"supported": 0.0, "unsupported": 0.0, "needs_human": 1.0}, + "score": 1.0, + "score_level": 1, + "score_name": "Needs a human", + "score_probabilities": {"0": 0.0, "1": 1.0, "2": 0.0}, + "expected_choice": None, + "expected_score_level": None, + "choice_correct": None, + "score_correct": None, + "latency_ms": 0, + "backend": "withheld", + "model": None, + "requested_model": None, + "worker": None, + "calls": [], + "note": "no_draft", + } + + +def _attach( + item: Mapping[str, Any], + decision: Mapping[str, Any], + *, + expected_choice: str | None, + expected_level: int | None, +) -> dict[str, Any]: + choice_correct = None if expected_choice is None else decision["choice"] == expected_choice + score_correct = None if expected_level is None else decision["score_level"] == expected_level + return { + "id": item["id"], + "choice": decision["choice"], + "probabilities": dict(decision["probabilities"]), + "score": decision["score"], + "score_level": decision["score_level"], + "score_name": decision["score_name"], + "score_probabilities": dict(decision["score_probabilities"]), + "expected_choice": expected_choice, + "expected_score_level": expected_level, + "choice_correct": choice_correct, + "score_correct": score_correct, + "latency_ms": decision["latency_ms"], + "backend": decision["backend"], + "model": decision["model"], + "requested_model": decision["requested_model"], + "worker": decision["worker"], + "calls": list(decision["calls"]), + } + + +def _stage_summary(records: list[dict[str, Any]], *, kept_key: str | None) -> dict[str, Any]: + labeled = [record for record in records if record["expected_choice"] is not None] + choice_correct = sum(1 for record in labeled if record["choice_correct"]) + score_labeled = [record for record in records if record["expected_score_level"] is not None] + score_correct = sum(1 for record in score_labeled if record["score_correct"]) + summary: dict[str, Any] = { + "total": len(records), + "labeled": len(labeled), + "choice_correct": choice_correct, + "score_correct": score_correct, + "choice_accuracy": _ratio(choice_correct, len(labeled)), + "score_accuracy": _ratio(score_correct, len(score_labeled)), + "records": records, + } + if kept_key is not None: + kept_ids = [record["id"] for record in records if record["choice"] == "keep"] + summary["kept"] = len(kept_ids) + summary["dropped"] = len(records) - len(kept_ids) + summary["kept_ids"] = kept_ids + return summary + + +def _ratio(correct: int, labeled: int) -> float | None: + if labeled == 0: + return None + return correct / labeled diff --git a/tests/test_content_workflow.py b/tests/test_content_workflow.py new file mode 100644 index 0000000..86d81bc --- /dev/null +++ b/tests/test_content_workflow.py @@ -0,0 +1,533 @@ +"""Content workflow: fixture ingest, human gates, offline rules, and the Omni client contract.""" + +from __future__ import annotations + +import contextlib +import json +import os +from io import StringIO +from pathlib import Path +from tempfile import TemporaryDirectory +from unittest import TestCase +from unittest.mock import patch + +import httpx + +import s1a.content.__main__ as content_cli +from s1a.content.ingest import ( + LIVE_X, + UnconfiguredXAdapter, + load_inputs, + load_jsonl, + load_rss_file, + load_rss_url, + load_x_fixture, + normalize_item, + parse_feed, + posts_to_items, +) +from s1a.content.latency import gate, omni_call_latencies, percentile +from s1a.content.omni_client import ( + OmniClient, + check_base_url, + decider_from_env, + require_worker, + worker_record, +) +from s1a.content.rules import SCREEN_CHOICES, RuleDecider, fact_state, lexical_fact, lexical_screen, screen_state +from s1a.content.workflow import GateRequired, run_workflow + +ROOT = Path(__file__).resolve().parents[1] +RECIPE = ROOT / "recipes" / "content-workflow" +ATOM = """ + + Desk + + Agent research note + + atom-1 + A research note about an agent. + + +""" + + +def _keep(ident: str, **extra: object) -> dict: + row = { + "id": ident, + "title": f"Model note {ident}", + "summary": "A model note for the agent desk.", + "source": "fixture", + "label": "keep", + } + row.update(extra) + return row + + +def _drop(ident: str) -> dict: + return { + "id": ident, + "title": "Rain expected overnight", + "summary": "Showers move in after midnight.", + "source": "fixture", + "label": "drop", + } + + +class TestIngest(TestCase): + def test_rss_file_atom_and_x_fixture_stay_offline(self) -> None: + rss = load_rss_file(RECIPE / "sample.rss") + self.assertEqual([row["id"] for row in rss], ["rss-keep-1", "rss-drop-1"]) + self.assertEqual(rss[0]["source"], "rss") + self.assertNotIn("label", rss[0]) + atom = parse_feed(ATOM, origin="atom") + self.assertEqual(atom[0]["id"], "atom-1") + self.assertEqual(atom[0]["url"], "https://example.invalid/atom-1") + posts = load_x_fixture(RECIPE / "x_posts.jsonl") + self.assertEqual([row["id"] for row in posts], ["x-local-1", "x-local-2"]) + self.assertEqual(posts[0]["source"], "x") + self.assertIn("agent", posts[0]["summary"]) + + def test_rss_url_uses_the_given_client_and_rejects_a_huge_body(self) -> None: + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(str(request.url)) + return httpx.Response(200, content=(RECIPE / "sample.rss").read_bytes()) + + client = httpx.Client(transport=httpx.MockTransport(handler)) + rows = load_rss_url("https://example.invalid/feed", client=client) + self.assertEqual(len(rows), 2) + self.assertEqual(seen, ["https://example.invalid/feed"]) + client.close() + + def huge(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"x" * (2_000_001)) + + with self.assertRaises(ValueError): + load_rss_url("https://example.invalid/big", client=httpx.Client(transport=httpx.MockTransport(huge))) + + def test_bad_jsonl_names_the_line_and_hides_the_text(self) -> None: + secret = "private-ticket-body" + with TemporaryDirectory() as tmp: + path = Path(tmp) / "items.jsonl" + path.write_text("{not-json " + secret + "\n", encoding="utf-8") + with self.assertRaises(ValueError) as caught: + load_jsonl(path) + self.assertIn("line 1", str(caught.exception)) + self.assertNotIn(secret, str(caught.exception)) + + def test_unknown_fields_and_the_x_adapter_do_not_travel(self) -> None: + item = normalize_item( + {"id": "a", "title": "Model", "summary": "A model.", "source": "fixture", "SECRET": "nope", "label": "keep"} + ) + self.assertNotIn("SECRET", item) + with self.assertRaises(ValueError) as caught: + UnconfiguredXAdapter().fetch() + self.assertEqual(str(caught.exception), LIVE_X) + self.assertNotIn("api.x.com", json.dumps(posts_to_items([{"id": "p", "text": "A model note."}]))) + + def test_labelled_sample_is_fifty_items_and_matches_the_lexical_rule(self) -> None: + rows = load_jsonl(RECIPE / "items.jsonl") + self.assertEqual(len(rows), 50) + self.assertEqual(sum(row["label"] == "keep" for row in rows), 32) + self.assertEqual(sum(row["label"] == "drop" for row in rows), 18) + drafts = [row for row in rows if row.get("draft")] + self.assertEqual(len(drafts), 8) + for row in rows: + self.assertEqual(lexical_screen(row), row["label"]) + self.assertTrue(row["url"].startswith("https://example.invalid/")) + for row in drafts: + self.assertEqual(lexical_fact(row["draft"]), row["draft"]["label"]) + self.assertEqual(lexical_screen({"title": "Supermodel cafe", "summary": "Downtown."}), "drop") + self.assertEqual(lexical_screen({"title": "Chipotle sauce", "summary": "Lunch."}), "drop") + self.assertEqual(lexical_screen({"title": "Agency hours", "summary": "Friday."}), "drop") + self.assertEqual(lexical_screen({"title": "A model card", "summary": "Research."}), "keep") + + +class TestGates(TestCase): + def _three(self) -> list[dict]: + return [_keep("a"), _keep("b"), _keep("c"), _drop("d")] + + def test_missing_pick_stops_before_fact_check_and_publishes_nothing(self) -> None: + class Spy(RuleDecider): + def __init__(self) -> None: + self.facts = 0 + + def fact_check(self, item: dict) -> dict: + self.facts += 1 + return super().fact_check(item) + + spy = Spy() + with self.assertRaises(GateRequired) as caught: + run_workflow(self._three(), decider=spy) + self.assertEqual(spy.facts, 0) + packet = caught.exception.packet + assert packet is not None + self.assertEqual(packet["stopped_at"], "human_pick") + self.assertEqual(packet["human_publish"]["executed"], []) + self.assertFalse(packet["auto_publish"]) + self.assertIsNone(packet["fact_check"]) + + def test_pick_must_be_three_to_five_kept_items(self) -> None: + rows = self._three() + with self.assertRaises(ValueError): + run_workflow(rows, decider=RuleDecider(), picks=["a", "b"], publish={"a": "hold", "b": "hold"}) + with self.assertRaises(ValueError): + run_workflow(rows, decider=RuleDecider(), picks=["a", "b", "d"], publish={}) + with self.assertRaises(GateRequired) as caught: + run_workflow(rows, decider=RuleDecider(), picks=["a", "b", "c"]) + packet = caught.exception.packet + assert packet is not None + self.assertEqual(packet["stopped_at"], "human_publish") + self.assertEqual([row["id"] for row in packet["fact_check"]["records"]], ["a", "b", "c"]) + self.assertEqual(packet["human_publish"]["executed"], []) + + def test_explicit_publish_is_recorded_and_not_executed(self) -> None: + report = run_workflow( + self._three(), + decider=RuleDecider(), + picks=["a", "b", "c"], + publish={"a": "publish", "b": "hold", "c": "publish"}, + ) + self.assertEqual(report["human_publish"]["decisions"]["a"], "publish") + self.assertEqual(report["human_publish"]["executed"], []) + self.assertFalse(report["auto_publish"]) + self.assertFalse(report["unsafe_skip"]) + self.assertEqual(report["stages"], ["ingest", "screen", "human_pick", "fact_check", "human_publish"]) + + def test_unsafe_skip_invents_a_hold_and_still_does_not_publish(self) -> None: + report = run_workflow(self._three(), decider=RuleDecider(), allow_unsafe_skip_gates=True) + self.assertTrue(report["unsafe_skip"]) + self.assertEqual(report["human_pick"]["source"], "unsafe-skip") + self.assertEqual(report["human_pick"]["ids"], ["a", "b", "c"]) + self.assertEqual(set(report["human_publish"]["decisions"].values()), {"hold"}) + self.assertEqual(report["human_publish"]["executed"], []) + self.assertFalse(report["auto_publish"]) + + def test_explicit_decisions_win_over_the_unsafe_flag_and_hooks(self) -> None: + def boom(_rows: list[dict]) -> list[str]: + raise AssertionError("hook should not run") + + report = run_workflow( + self._three(), + decider=RuleDecider(), + picks=["a", "b", "c"], + publish={"a": "hold", "b": "hold", "c": "hold"}, + pick_hook=boom, + allow_unsafe_skip_gates=True, + ) + self.assertEqual(report["human_pick"]["source"], "explicit") + self.assertFalse(report["unsafe_skip"]) + + def test_a_hook_is_a_real_gate_and_a_missing_draft_is_not_invented(self) -> None: + def pick(kept: list[dict]) -> list[str]: + self.assertNotIn("label", kept[0]) + return [row["id"] for row in kept] + + def publish(checked: list[dict]) -> dict[str, str]: + return {row["id"]: "hold" for row in checked} + + class Spy(RuleDecider): + def __init__(self) -> None: + self.fact_ids: list[str] = [] + + def fact_check(self, item: dict) -> dict: + self.fact_ids.append(item["id"]) + if "label" in item or "label" in item["draft"]: + raise AssertionError("fact-check saw a label") + return super().fact_check(item) + + draft = { + "claims": ["The model note is ready."], + "sources": [{"id": "s", "text": "The model note is ready for the desk."}], + "label": "supported", + } + rows = [_keep("a", draft=draft), _keep("b"), _drop("c")] + spy = Spy() + report = run_workflow(rows, decider=spy, pick_hook=pick, publish_hook=publish) + self.assertEqual(report["human_pick"]["source"], "hook") + self.assertEqual(spy.fact_ids, ["a"]) + notes = {row["id"]: row.get("note") for row in report["fact_check"]["records"]} + self.assertEqual(notes["b"], "no_draft") + self.assertEqual(report["fact_check"]["records"][0]["choice_correct"], True) + self.assertNotIn("SECRET", json.dumps(report)) + + def test_screen_does_not_receive_labels_or_drafts(self) -> None: + seen = [] + + class Spy(RuleDecider): + def screen(self, item: dict) -> dict: + seen.append(item) + return super().screen(item) + + draft = { + "claims": ["StartLux Instant 4B is the default demo worker for this desk."], + "sources": [{"id": "s", "text": "The default demo worker for this desk is StartLux Instant 4B."}], + } + rows = [_keep("a", draft=draft, SECRET="hidden"), _drop("b"), _keep("c")] + run_workflow( + rows, + decider=Spy(), + picks=["a", "c"], + publish={"a": "hold", "c": "hold"}, + ) + self.assertTrue(seen) + for item in seen: + self.assertNotIn("label", item) + self.assertNotIn("draft", item) + self.assertNotIn("SECRET", item) + self.assertNotIn("StartLux", screen_state(seen[0])) + + +class TestOmniClient(TestCase): + def test_clef_is_refused_and_pplx_needs_its_own_url(self) -> None: + with self.assertRaises(ValueError) as caught: + require_worker("clef", fact_check=True) + self.assertIn("165", str(caught.exception)) + with self.assertRaises(ValueError): + decider_from_env({"OMNI_BASE_URL": "http://127.0.0.1:8000"}, fact_check_worker="pplx") + with self.assertRaises(ValueError): + check_base_url("http://127.0.0.1:8000/v1/systemone") + with self.assertRaises(ValueError): + require_worker("pplx", fact_check=False) + + def test_short_prefill_posts_one_question_and_hides_the_error_body(self) -> None: + bodies = [] + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/health": + return httpx.Response(200, json={"status": "ok"}) + body = json.loads(request.content) + bodies.append((str(request.url), body)) + self.assertEqual(len(body["questions"]), 1) + self.assertLessEqual(len(body["state"]), 1500) + self.assertNotIn("super-secret-key", request.content.decode()) + self.assertNotIn("label", body) + question_id, question = next(iter(body["questions"].items())) + if "8001" in str(request.url): + self.assertEqual(body["thinking"], {"enabled": False}) + self.assertNotIn("think_mode", body) + self.assertEqual(body["model"], "perplexity-ai/pplx-decider-v1.1-27b") + choice, level, keys = "supported", 2, list(question["criteria"]) + else: + self.assertEqual(body["think_mode"], "never") + self.assertNotIn("thinking", body) + self.assertEqual(body["model"], "StartLux-Decision-v1.1-4B-Instant") + choice, level, keys = "keep", 1, list(question["criteria"]) + if question["type"] == "choice": + answer = { + "type": "choice", + "choice": choice, + "probabilities": {key: (1.0 if key == choice else 0.0) for key in keys}, + } + else: + answer = { + "type": "score", + "score": float(level), + "probabilities": {str(i): (1.0 if i == level else 0.0) for i in range(len(question["criteria"]))}, + } + return httpx.Response(200, json={"model": body["model"], "answers": {question_id: answer}}) + + env = { + "OMNI_BASE_URL": "http://127.0.0.1:8000", + "OMNI_FACTCHECK_BASE_URL": "http://127.0.0.1:8001", + "OMNI_API_KEY": "super-secret-key", + } + decider = decider_from_env(env, fact_check_worker="pplx", handler=handler) + self.assertEqual(decider.screen_client.health()["status"], "ok") + draft = { + "claims": ["StartLux Instant 4B is the default demo worker for this desk."], + "sources": [{"id": "s", "text": "The default demo worker for this desk is StartLux Instant 4B."}], + "label": "supported", + } + rows = [ + _keep("a", draft=draft, SECRET="super-secret-key"), + _drop("b"), + _keep("c"), + ] + try: + report = run_workflow( + rows, + decider=decider, + picks=["a", "b", "c"], + publish={"a": "publish", "b": "hold", "c": "hold"}, + workers={ + "screen": worker_record( + decider.screen_client.profile, decider.screen_client.base_url, decider.screen_client.model + ), + "fact_check": worker_record( + decider.fact_client.profile, decider.fact_client.base_url, decider.fact_client.model + ), + }, + ) + finally: + decider.close() + self.assertEqual(report["backend"], "omni") + self.assertEqual(report["workers"]["screen"]["pr"], 164) + self.assertEqual(report["workers"]["fact_check"]["pr"], 166) + self.assertFalse(report["workers"]["fact_check"]["hot_path"]) + self.assertNotIn("super-secret-key", json.dumps(report)) + self.assertEqual(report["human_publish"]["executed"], []) + self.assertEqual(report["human_publish"]["decisions"]["a"], "publish") + self.assertFalse(report["latency_gate"]["attached"]) + screen_calls = [url for url, _ in bodies if ":8000/" in url and url.endswith("/v1/systemone")] + fact_calls = [url for url, _ in bodies if ":8001/" in url] + self.assertEqual(len(screen_calls), 6) + self.assertEqual(len(fact_calls), 2) + self.assertEqual(report["screen"]["choice_correct"], 2) + self.assertEqual(report["fact_check"]["records"][0]["choice"], "supported") + self.assertEqual(report["fact_check"]["records"][1]["note"], "no_draft") + self.assertTrue(report["fact_check"]["records"][0]["calls"]) + + def test_a_failed_status_does_not_echo_the_body(self) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(500, json={"detail": "SECRET-BODY"}) + + client = OmniClient( + "http://127.0.0.1:8000", + profile=require_worker("startlux", fact_check=False), + transport=httpx.MockTransport(handler), + ) + try: + with self.assertRaises(RuntimeError) as caught: + client.ask( + "Title: Model\nSummary: A model.\nSource: fixture", + "screen", + {"type": "choice", "instructions": "Keep or drop.", "criteria": dict(SCREEN_CHOICES)}, + ) + finally: + client.close() + self.assertNotIn("SECRET-BODY", str(caught.exception)) + self.assertIn("HTTP 500", str(caught.exception)) + + +class TestWorkflowFixture(TestCase): + def test_recipe_files_walk_all_five_stages_on_the_rule_backend(self) -> None: + rows = load_jsonl(RECIPE / "items.jsonl") + picks = json.loads((RECIPE / "picks.json").read_text(encoding="utf-8"))["ids"] + publish = json.loads((RECIPE / "publish.json").read_text(encoding="utf-8"))["decisions"] + report = run_workflow(rows, decider=RuleDecider(), picks=picks, publish=publish) + self.assertEqual(report["screen"]["total"], 50) + self.assertEqual(report["screen"]["kept"], 32) + self.assertEqual(report["screen"]["choice_accuracy"], 1.0) + self.assertEqual(report["screen"]["score_accuracy"], 1.0) + self.assertEqual(report["human_pick"]["ids"], ["n01", "n02", "n03", "n04"]) + self.assertEqual( + [row["choice"] for row in report["fact_check"]["records"]], + ["supported", "unsupported", "needs_human", "supported"], + ) + self.assertEqual(report["fact_check"]["choice_accuracy"], 1.0) + self.assertEqual(report["fact_check"]["score_accuracy"], 1.0) + self.assertEqual(report["human_publish"]["decisions"], publish) + self.assertIn("publish", report["human_publish"]["decisions"].values()) + self.assertEqual(report["human_publish"]["executed"], []) + self.assertFalse(report["auto_publish"]) + self.assertEqual(omni_call_latencies(report), []) + state = fact_state(rows[0]) + self.assertNotIn("Label:", state) + self.assertLessEqual(len(state), 1500) + + def test_local_rss_and_x_fixtures_can_be_combined_without_a_network(self) -> None: + items = load_inputs(rss_files=[RECIPE / "sample.rss"], x_files=[RECIPE / "x_posts.jsonl"]) + report = run_workflow(items, decider=RuleDecider(), allow_unsafe_skip_gates=True) + self.assertEqual(report["ingest"]["count"], 4) + self.assertEqual(report["ingest"]["sources"], {"rss": 2, "x": 2, "fixture": 0}) + self.assertEqual(report["screen"]["kept"], 2) + self.assertEqual(report["human_publish"]["executed"], []) + self.assertTrue(report["unsafe_skip"]) + self.assertEqual(set(report["human_publish"]["decisions"].values()), {"hold"}) + + +class TestCli(TestCase): + def test_recipe_command_and_the_latency_hook(self) -> None: + out = self._run( + [ + "run", + "--items", + str(RECIPE / "items.jsonl"), + "--picks", + str(RECIPE / "picks.json"), + "--publish", + str(RECIPE / "publish.json"), + "--backend", + "rule", + ] + ) + self.assertEqual(out["code"], 0, out["err"]) + report = json.loads(out["stdout"]) + self.assertEqual(report["screen"]["choice_correct"], 50) + self.assertEqual(report["human_publish"]["executed"], []) + self.assertFalse(report["latency_gate"]["attached"]) + from tempfile import TemporaryDirectory + + with TemporaryDirectory() as tmp: + path = Path(tmp) / "report.json" + path.write_text(out["stdout"], encoding="utf-8") + idle = self._run(["latency-report", "--report", str(path)]) + self.assertEqual(idle["code"], 0, idle["err"]) + self.assertFalse(json.loads(idle["stdout"])["enforced"]) + blocked = self._run(["latency-report", "--report", str(path), "--p95-max-ms", "10", "--min-samples", "100"]) + self.assertEqual(blocked["code"], 1) + body = json.loads(blocked["stdout"]) + self.assertFalse(body["passed"]) + self.assertEqual(body["summary"]["n"], 0) + + def test_gates_cannot_be_skipped_without_the_unsafe_flag(self) -> None: + refused = self._run(["run", "--items", str(RECIPE / "items.jsonl"), "--backend", "rule"]) + self.assertEqual(refused["code"], 2) + self.assertIn("human pick gate", refused["err"]) + self.assertIn("allow-unsafe-skip-gates", refused["err"]) + skipped = self._run( + ["run", "--items", str(RECIPE / "items.jsonl"), "--backend", "rule", "--allow-unsafe-skip-gates"] + ) + self.assertEqual(skipped["code"], 0, skipped["err"]) + report = json.loads(skipped["stdout"]) + self.assertTrue(report["unsafe_skip"]) + self.assertEqual(report["human_publish"]["executed"], []) + self.assertEqual(set(report["human_publish"]["decisions"].values()), {"hold"}) + + def test_omni_without_a_worker_fails_before_any_publish(self) -> None: + refused = self._run( + [ + "run", + "--rss", + str(RECIPE / "sample.rss"), + "--picks", + str(RECIPE / "picks.json"), + "--publish", + str(RECIPE / "publish.json"), + "--backend", + "omni", + ], + env={"OMNI_BASE_URL": "http://127.0.0.1:9", "OMNI_API_KEY": "super-secret-key", "OMNI_TIMEOUT_S": "0.2"}, + ) + self.assertEqual(refused["code"], 1) + self.assertNotIn("super-secret-key", refused["err"]) + self.assertNotIn("publish", refused["stdout"]) + + def _run(self, argv: list[str], env: dict[str, str] | None = None) -> dict[str, str | int]: + merged = {"OMNI_BASE_URL": "", "OMNI_FACTCHECK_BASE_URL": "", "OMNI_API_KEY": "", "OMNI_TIMEOUT_S": ""} + if env: + merged.update(env) + stdout, stderr = StringIO(), StringIO() + with ( + patch.dict(os.environ, merged, clear=False), + patch.object(content_cli.sys.stdin, "isatty", return_value=False), + contextlib.redirect_stdout(stdout), + contextlib.redirect_stderr(stderr), + ): + code = content_cli.main(argv) + return {"code": code, "stdout": stdout.getvalue(), "err": stderr.getvalue()} + + +class TestLatencyMath(TestCase): + def test_nearest_rank_and_an_unattached_gate(self) -> None: + self.assertEqual(percentile([1, 2, 3, 4, 5], 50), 3) + self.assertEqual(percentile([1, 2, 3, 4, 5], 95), 5) + idle = gate([1.0, 2.0]) + self.assertFalse(idle["enforced"]) + self.assertIsNone(idle["passed"]) + closed = gate([], p95_max_ms=10, min_samples=5) + self.assertFalse(closed["passed"]) + self.assertIn("Public231 C1", closed["procedure"])