diff --git a/docs/api.md b/docs/api.md index f6c29a880..243c253f4 100644 --- a/docs/api.md +++ b/docs/api.md @@ -70,7 +70,7 @@ REST API at `/api/v1/`. Key endpoints: - `PATCH/DELETE /workspaces/{ws}/reminders/{id}`, `POST /workspaces/{ws}/reminders/{id}/ack` — re-arm (clears both fire marks), disarm, acknowledge. Permission is the ITEM's; the reminder has no separate owner - `POST /workspaces/{ws}/items/{slug}/claim` and `/release`: the item execution lease (#1221). Body `{holder?, worker_id?, ttl_seconds?}` (ttl on claim only: default 900, 1 to 86400, else 400). Needs edit permission and a signed-in user. A re-claim by the live holder extends the expiry and keeps `acquired_at` (the heartbeat). Release is idempotent: an absent or expired lease answers `released: false`. Expiry is absence on every read, with no reaper. **Identity (TASK-3613):** a lease is held under a WORKER ID plus the account. `worker_id` is `"wk_"` + 32 lowercase hex (128 random bits) that the client mints once per agent session (the CLI stores it in its session registry and sends it as `X-Pad-Worker`); a malformed one is `400 invalid_worker_id`. It comes from the body, else `X-Pad-Worker`, else a `?worker_id=` query, else, for a remote MCP call, an id derived from the account and the MCP session. A lease with a worker id is the caller's iff the worker id AND the account match; `holder` (default: the `X-Pad-Agent` name, else the email) is then display only. A lease without one (a name-only caller, or one claimed before migration 142) keeps the older rule: label plus account, with an unrecorded account honoured on the label alone (BUG-3341). A worker refreshing its own name-only lease binds it. A name-only caller never matches a worker-bound lease, even under the same label: that is refused `409 lease_held`, and the message says how to get a worker id. A contended claim or release answers `409 lease_held` with `details: {ref, holder, worker?, acquired_at, expires_at}`, where `worker` is the SHORT id (`wk_3f2a`). The full id appears only in the claim's own success answer (`worker_id`, for resume); reads (`item.lease`, list decoration) show `holder` and the short `worker`. `GET /next` and the dashboard's `suggested_next` leave out items whose live lease a claim by the same identity would lose, by the same predicate (`store.leaseOurs`). - `GET/POST /workspaces/{ws}/items/{slug}/verifications` — verification evidence tied to a revision (TASK-3617). One row is one verdict (`pass` / `fail` / `cant_tell`) on one acceptance criterion (`AC-N`) of the item that CARRIES the criteria, optionally at a `code_revision` (full 40-hex SHA) and naming the item that did the work (`via`). POST takes `{run_id?, code_revision?, via?, results: [{criterion, verdict, evidence?, evidence_url?}]}` (at most 100 results; evidence at most 2048 bytes; `evidence_url` an absolute http(s) URL) and is atomic: one bad result refuses the batch. A criterion that is missing, still a placeholder, or appears more than once in the BODY is `400 unknown_criterion`, with `details.criteria` listing the valid ids; a criterion named twice in one REQUEST is a malformed request, `400 validation_error`. A `via` that does not resolve in the workspace, or names an item the caller cannot see, is one answer: `400 unknown_via`. The write is guarded by the item's seq: the criterion hashes are stored against the body they were parsed from, and an item that keeps changing during the write answers `409 item_changed`. Needs edit permission on the item; the read follows item visibility. Writes log a `verified` activity (metadata: `run_id`, a verdict count, never evidence text) and publish `item_updated`. GET returns `{ref, criteria, verifications}`, the newest verdict per criterion in body order, or the whole history with `?all=1`. Each row's `state` is computed at read time from the criterion, never from the item's seq: `criterion_removed` (no longer a real criterion), `criterion_changed` (its text changed since the verdict; whitespace is collapsed first, so re-wrapping is not a change), `superseded` (a newer verdict exists), else `current`. `possibly_stale: {via: "PR #N", merged_sha, merged_at}` is ADVISORY and never changes `state`: a merged `github_pr` on the item, or on its visible `via` item, merged after the verdict at a different SHA. It needs `github_pr.merge_sha` / `merged_at`, which `pad github link` and `pad reconcile` now record (`merge_sha` must be a full 40-hex SHA, `merged_at` RFC3339, or the PR write is `400`). Rows are append-only and travel with a workspace export/import (so `pad db migrate-to-pg` carries them), verbatim except for the remapped item and via ids; the verifier's user id does not travel, and a malformed bundle row is dropped rather than failing the import. They survive an archive and restore, are removed by a hard delete of the item, are not cloned by a copy (the copy's dropped bucket does not report them yet), and outlive a deleted `via` item (whose ref is then omitted) -- `GET /workspaces/{ws}/items/{slug}/decisions` — the item's latest typed-decision answer per question (PLAN-3114 / TASK-3117), each with `current: true` only when it was computed from the item's PRESENT state (a sha256 of the exact state bytes sent to the provider — title, collection, fields, bounded body, and per set: the last 10 comments for `attention`, and the parent and blocking links and NO comments for `conventions`) AND for the question as registered now under the model pinned now — rewording a question or changing the model re-asks. A third condition since TASK-3118: the set must still APPLY to the item (its collection is covered and the set's eligibility predicate accepts the item as it stands). A collection schema edit can make an item terminal without changing a byte of its state, and its answers then read `current: false`. Evaluation is asynchronous: item create / update / move and comment create enqueue a coalesced `decision_jobs` row in their own transaction and a server tick does the provider call, so nothing on a write path waits on the network. A re-ask is debounced (TASK-3916): a write within an hour of the item's latest evaluation for a set owes a job (`decision_jobs.not_before`) that the tick does not claim until that hour has passed, and later writes in the window coalesce into it without moving it. A first evaluation is never delayed, and a per-comment set has no window. System fan-out rewrites (title-rename cascade, field migration, attachment remap) deliberately do NOT enqueue — their items read `current: false` until the next direct write — and neither does workspace import, whose items have no decisions until their next direct write. With no provider configured the list is empty and no job row is ever written. The single-item GET (and `pad item show --format json` / `--agent`) carries the same list as `decisions` when non-empty. Two sets are registered. `attention` (TASK-3118): one Noul, `needs_human_decision`, asked only about OPEN items in NON-SYSTEM collections. Its `blocked` and `waiting_on_external` Nouls were retired in TASK-3915 / TASK-3924 (the dependency graph owns blocked). Their stored answers stay in the list with `current: false`, because no registered question carries their keys. A job owed for a terminal item is dropped at the tick without a provider call, and reopening the item re-enqueues it. `conventions` (TASK-3119): one Noul per active `always` convention whose `decision_check` is not `off` and whose body is at most 4 KB, keyed `conv:`, asked about items in non-system collections, at most 12 per provider call; the web shows a "Possibly breaks CONVE-N" chip at p >= 0.9 and nothing below it. Its state carries no comment trail (U2a), so a comment write changes nothing it reads and costs it no call. `conventions_comments` (U2b), which asked the same questions about each COMMENT alone keyed `conv:@`, is no longer registered (TASK-3924: 42% of provider calls for almost no chip-level answers). Its stored rows read `current: false`, and the web shows no per-comment chip. Each evaluation that reaches the provider logs one `decision evaluation` line with its calls and tokens +- `GET /workspaces/{ws}/items/{slug}/decisions` — the item's latest typed-decision answer per question (PLAN-3114 / TASK-3117), each with `current: true` only when it was computed from the item's PRESENT state (a sha256 of the exact state bytes sent to the provider — title, collection, fields, bounded body, and per set: the last 10 comments for `attention`, and the parent and blocking links and NO comments for `conventions`) AND for the question as registered now under the model pinned now — rewording a question or changing the model re-asks. A third condition since TASK-3118: the set must still APPLY to the item (its collection is covered and the set's eligibility predicate accepts the item as it stands). A collection schema edit can make an item terminal without changing a byte of its state, and its answers then read `current: false`. Evaluation is asynchronous: item create / update / move and comment create enqueue a coalesced `decision_jobs` row in their own transaction and a server tick does the provider call, so nothing on a write path waits on the network. A re-ask is debounced (TASK-3916): a write within an hour of the item's latest evaluation for a set owes a job (`decision_jobs.not_before`) that the tick does not claim until that hour has passed, and later writes in the window coalesce into it without moving it. A first evaluation is never delayed, and a per-comment set has no window. System fan-out rewrites (title-rename cascade, field migration, attachment remap) deliberately do NOT enqueue — their items read `current: false` until the next direct write — and neither does workspace import, whose items have no decisions until their next direct write. With no provider configured the list is empty and no job row is written, with one exception: if decisions are disabled while the one-shot effort backfill (TASK-3921) is writing a chunk, that chunk may still commit, and its jobs wait until a provider is configured again. The single-item GET (and `pad item show --format json` / `--agent`) carries the same list as `decisions` when non-empty, with the staged keys (`effort`, `effort_medium`, `high_stakes`) omitted; this endpoint returns them. Two sets are registered. `attention` (TASK-3118): four questions in one call, asked only about OPEN items in NON-SYSTEM collections. `needs_human_decision` (a Noul) is the one surfaced anywhere. `effort` (a Score over four levels), `effort_medium` (a Choice: desk, device, other_person, unclear) and `high_stakes` (a Noul) are STAGED before their reader (TASK-3959, IDEA-3906 B4b): they are stored and returned by this endpoint, but no dashboard entry, chip, catch-up or item brief shows or counts them. A one-shot backfill owes the attention set to every item that has a `needs_human_decision` answer and no `effort` answer and that the attention set applies to now (open, in a non-system collection that tracks work). It owes them in chunks of one tick pass (20 by default), each chunk only after the previous chunk's own jobs have been taken (no job of it is claimable right now), polling at the tick interval, so a live re-ask interleaves with the backlog rather than waiting behind all of it; a job held by the re-ask debounce cannot be claimed yet and does not hold the next chunk, and because those come due later, a few of them may become claimable together ahead of a live re-ask, once. The marker is written once the last chunk is owed, without waiting for it to drain. It is enqueued when the provider is configured (boot, or a later admin enable), runs on the tick under the re-ask debounce, makes no provider call itself, and is gated by the `effort_decisions_backfilled` platform setting, written only after every job was owed. Its `blocked` and `waiting_on_external` Nouls were retired in TASK-3915 / TASK-3924 (the dependency graph owns blocked). Their stored answers stay in the list with `current: false`, because no registered question carries their keys. A job owed for a terminal item is dropped at the tick without a provider call, and reopening the item re-enqueues it. `conventions` (TASK-3119): one Noul per active `always` convention whose `decision_check` is not `off` and whose body is at most 4 KB, keyed `conv:`, asked about items in non-system collections, at most 12 per provider call; the web shows a "Possibly breaks CONVE-N" chip at p >= 0.9 and nothing below it. Its state carries no comment trail (U2a), so a comment write changes nothing it reads and costs it no call. `conventions_comments` (U2b), which asked the same questions about each COMMENT alone keyed `conv:@`, is no longer registered (TASK-3924: 42% of provider calls for almost no chip-level answers). Its stored rows read `current: false`, and the web shows no per-comment chip. Each evaluation that reaches the provider logs one `decision evaluation` line with its calls and tokens - `POST /workspaces/{ws}/playbooks/{ref}/run` — bind a playbook's declared arguments and return its body (`pad playbook run`, `pad_playbook.action: run`). Side-effect-free: nothing is stored. So it is the ONE POST a read-scoped token (`read` / `pad:read`, PAT or OAuth) may make (TASK-2863). That is an exact-path allowlist in `tokenScopeAllows`; both variable segments must be plain names (RFC 3986 unreserved characters). `/playbooks/match` stays refused for read tokens, as do every write and the item `claim` / `release` routes - `POST /workspaces/{ws}/playbooks/match {text}` — which ACTIVE playbook (or `none`) free text asks for, answered by the decision provider synchronously (TASK-3120; `pad playbook match -- ""`, `pad_playbook.action: match`). 404 `decision_provider_unavailable` with no provider, so callers fall back to slug/trigger routing. Each call that reaches the provider spends from its own bucket, 30/min per user, burst 5, charged immediately before the provider call, so an unconfigured 404, a 400 for the text, a too-many-playbooks 400 or a zero-playbook answer consumes nothing (a request-too-large 400 is raised inside that call and has spent its token); over it is `429` with `Retry-After` (TASK-3141). No env knob by design: tuning belongs in the instance-admin decision-provider setting - `GET /workspaces/{ws}/dashboard` — computed project overview (active items, plans, attention, blockers). Also carries `pending_reminders`: fired-but-unacknowledged reminders, which is the delivery path on any instance with no webhook configured. With a decision provider, `attention` also carries `needs_human` entries (the item's latest `needs_human_decision` answer is >= 0.7). `blocked` entries come from the dependency graph only; the text-derived kind was retired in TASK-3915. Only CURRENT answers count, the rule the decisions endpoint applies (the item's present state, the question as registered now, the model pinned now), with one exception (TASK-3915): an item edited or commented since its last evaluation KEEPS its entry while the re-ask is pending, that is while a job is owed for it with no recorded error and the item is still open. It drops off when no job is owed or the job has failed, and the re-ask's answer replaces it. The state check costs one item-plus-trail read per ABOVE-threshold item, never per workspace item. A failing provider (any owed `attention` job with a recorded error) marks the section `attention.decisions` in `degraded_sections`. `pad project stale` filters on type, so it reports graph-`blocked` entries and not `needs_human` @@ -87,7 +87,7 @@ REST API at `/api/v1/`. Key endpoints: - `GET /api/v1/s/{token}` — public share link resolution. For a COLLECTION share the items are **bounded and paged** (BUG-3792): `limit` (default and cap 1000) and `offset` query parameters, and the body carries `has_more` and `next_offset` (the headers are set too). The web share page follows `next_offset` until `has_more` is false and merges the pages' items and `attachment_refs` - `GET /workspaces/{ws}/items/{slug}/progress` — child item completion progress - `GET/POST /workspaces/{ws}/items/{slug}/links` — item relationships (blocks/blocked-by, parent/child) -- `GET /workspaces/{ws}/items/{slug}/context?intent=&role=&budget=` (PLAN-3598) — one deterministic brief for working an item: `item` (fields, body, seq), `parents` (the visible ancestor chain, nearest first, each with `acceptance_criteria` parsed by `internal/acceptance` and the nearest one's body), `conventions` (active ones whose trigger is `always` or the intent's, with bodies, sorted must → should → nice-to-have then ref), `playbooks` (metadata only), `deps` (`blocks` / `blocked_by` with statuses), `decisions` (TASK-3618: only the answers their set would surface, i.e. the question is current and the Noul reaches the set's threshold, `decision.AttentionThreshold` or `decision.ConventionThreshold`; each carries `current`, and a surfaced answer whose item has moved on stays, with `current: false` and a `stale_decision` conflict. The rest are only counted: `decisions_below_threshold`, and `decisions_stale_question` for answers to a question no longer asked as it was), `links` (other link types, ref/title/status, no bodies), `omitted`, and `conflicts`. Every part carries its ref and seq. **Intent** resolves against the conventions collection's declared trigger options: `on-` if declared, else ``; neither is 400 `unknown_intent` with `details.valid_intents`. No intent means `always` only. A caller who cannot see the conventions collection gets no validation and no list (the intent is taken as `on-`), so the 400 cannot reveal its declared triggers. **Role** must be a role slug in the workspace (400 `unknown_role`); with one, a convention scoped to another role (by its agent-role column, the one bootstrap labels with) is left out, and without one every role's are included, as bootstrap does. **Budget** (characters of serialized parts, default 40000, capped at 400000) admits whole parts in a fixed order: parents, conventions, playbooks, deps, decisions, links. The first part that does not fit and every part after it are listed in `omitted` (`part`, `ref`, `seq`, `size`, `reason: "budget"`), so a later small part never slips in after a refused one. The item always ships, even past the budget. **Visibility:** the item uses the single-item read check, so an archived item is served read-only to a caller who can read it, exactly as the single-item GET serves it, and an unreadable item is the same 404 as a missing one; every ancestor, dependency and linked item uses `itemVisibleFor` plus the app token's collection ceiling and live rows only; conventions and playbooks use bootstrap's visibility sets. A part the caller cannot see is absent and NOT counted in `omitted`. **`conflicts`** is structural only (no semantic comparison of rules): `open_blocker`, `open_under_closed_parent`, `stale_decision`, and `role_and_global_convention` (a role-scoped and a global convention matching the same trigger under the same title) +- `GET /workspaces/{ws}/items/{slug}/context?intent=&role=&budget=` (PLAN-3598) — one deterministic brief for working an item: `item` (fields, body, seq), `parents` (the visible ancestor chain, nearest first, each with `acceptance_criteria` parsed by `internal/acceptance` and the nearest one's body), `conventions` (active ones whose trigger is `always` or the intent's, with bodies, sorted must → should → nice-to-have then ref), `playbooks` (metadata only), `deps` (`blocks` / `blocked_by` with statuses), `decisions` (TASK-3618: only the answers their set would surface (the staged `attention` keys never are, nor are they counted), i.e. the question is current and the Noul reaches the set's threshold, `decision.AttentionThreshold` or `decision.ConventionThreshold`; each carries `current`, and a surfaced answer whose item has moved on stays, with `current: false` and a `stale_decision` conflict. The rest are only counted: `decisions_below_threshold`, and `decisions_stale_question` for answers to a question no longer asked as it was), `links` (other link types, ref/title/status, no bodies), `omitted`, and `conflicts`. Every part carries its ref and seq. **Intent** resolves against the conventions collection's declared trigger options: `on-` if declared, else ``; neither is 400 `unknown_intent` with `details.valid_intents`. No intent means `always` only. A caller who cannot see the conventions collection gets no validation and no list (the intent is taken as `on-`), so the 400 cannot reveal its declared triggers. **Role** must be a role slug in the workspace (400 `unknown_role`); with one, a convention scoped to another role (by its agent-role column, the one bootstrap labels with) is left out, and without one every role's are included, as bootstrap does. **Budget** (characters of serialized parts, default 40000, capped at 400000) admits whole parts in a fixed order: parents, conventions, playbooks, deps, decisions, links. The first part that does not fit and every part after it are listed in `omitted` (`part`, `ref`, `seq`, `size`, `reason: "budget"`), so a later small part never slips in after a refused one. The item always ships, even past the budget. **Visibility:** the item uses the single-item read check, so an archived item is served read-only to a caller who can read it, exactly as the single-item GET serves it, and an unreadable item is the same 404 as a missing one; every ancestor, dependency and linked item uses `itemVisibleFor` plus the app token's collection ceiling and live rows only; conventions and playbooks use bootstrap's visibility sets. A part the caller cannot see is absent and NOT counted in `omitted`. **`conflicts`** is structural only (no semantic comparison of rules): `open_blocker`, `open_under_closed_parent`, `stale_decision`, and `role_and_global_convention` (a role-scoped and a global convention matching the same trigger under the same title) - `GET /search?q=query&workspace=slug` — full-text search. A `collection` filter resolves like a collection in the item routes (BUG-2659): exact slug first, then the singular/alias fallback, and an archived collection still claims its name. It resolves once per workspace in scope, so `collection=task` means `task` where one exists and `tasks` where it does not. `GET /server/capabilities` advertises it as `search_collection_resolution`; the CLI sends the name as typed only to a server that does, and expands the shorthand itself otherwise. `include_archived=true` (TASK-2864) adds soft-deleted items on every path (full text, ref and number lookups, total, facets), each carrying `deleted_at`. The permission filter is unchanged, and an individual item grant names live items only, so a guest sees an archived item only through a collection grant. The web search box's `is:archived` token sends it; not on the CLI or MCP - `GET /api/v1/events?workspace=slug` — SSE real-time events (workspace-scoped) - `GET /api/v1/events/stream` — SSE watch/push notifications (USER-scoped, spans every workspace the caller belongs to; backs `pad watch --stream`). With `?access=true` (TASK-3272) the stream also carries `workspace_access_changed` addressed to the caller — `{workspace_id, change}` with change `gained` / `lost` / `deleted` / `restored` / `purged` — published after every door that changes whether an EXISTING account reaches a workspace (signup-time doors publish nothing: the new account holds no stream yet). Opt-in because an installed `pad watch` prints any kind it does not know as an item line into an agent session. It is a refetch hint: correctness lives on the read-filtered `GET /me/workspace-tabs`, and a duplicate or missed event under concurrent grant writes is benign. Such a stream counts against `PAD_SSE_MAX_PER_USER` like any other. A push event carries `push_id` (TASK-3585, omitted on every other kind): the same id on every publish of one keyed push, so a consumer drops one it has seen (`pad watch --stream --for-session` keeps ~256 ids for 10 minutes); a monitor that predates it is at-least-once diff --git a/internal/appstore/boundary_check_test.go b/internal/appstore/boundary_check_test.go index d7d44ad12..3c2011778 100644 --- a/internal/appstore/boundary_check_test.go +++ b/internal/appstore/boundary_check_test.go @@ -71,6 +71,7 @@ var allowedStoreMethodsInsideDoor = map[string]string{ "outboxRowCap": "reads in-memory config; no SQL", "outboxClaimableRowCap": "reads in-memory config; no SQL", "enqueueDecisionJobsTx": "INSERT ... ON CONFLICT DO UPDATE on decision_jobs for the item it is given; no-op without a provider. Reads MAX(item_decisions.evaluated_at) for the item when its set has a debounce window (TASK-3916)", + "owedDecisionJobTx": "INSERT ... ON CONFLICT DO UPDATE on decision_jobs for the item and set it is given (the per-set body of enqueueDecisionJobsTx, TASK-3921); reads MAX(item_decisions.evaluated_at) for the item when the set has a debounce window", "decisionSetResolver": "loads an in-memory pointer; no SQL", "decisionDebounceFunc": "loads an in-memory pointer; no SQL (TASK-3916)", "createActivityQ": "INSERT INTO activities on the executor it is given (the fence's tx)", @@ -118,6 +119,9 @@ var allowedFuncValues = map[string]map[string]string{ "resolve": "the store's DecisionSetResolver; production installs decision.Registry.SetsFor, an in-memory lookup that runs no SQL", "debounce": "the store's DecisionDebounce (TASK-3916), passed to decisionWindowFor; production installs decision.Runner.debounceFor, a registry lookup that runs no SQL", }, + "(*github.com/PerpetualSoftware/pad/internal/store.Store).owedDecisionJobTx": { + "debounce": "the store's DecisionDebounce (TASK-3916), passed to decisionWindowFor; production installs decision.Runner.debounceFor, a registry lookup that runs no SQL", + }, "github.com/PerpetualSoftware/pad/internal/store.decisionWindowFor": { "debounce": "the store's DecisionDebounce (TASK-3916); production installs decision.Runner.debounceFor, a registry lookup that runs no SQL", }, diff --git a/internal/decision/attention.go b/internal/decision/attention.go index 20dc72742..a2c42a748 100644 --- a/internal/decision/attention.go +++ b/internal/decision/attention.go @@ -8,9 +8,11 @@ import ( "github.com/PerpetualSoftware/pad/internal/models" ) -// The `attention` question set (PLAN-3114 unit 3, TASK-3118): one Noul over +// The `attention` question set (PLAN-3114 unit 3, TASK-3118): four questions over // an open item and its recent trail, read by the dashboard's attention // section and shown as a chip on the item page. +// One of the four, needs_human_decision, is read by both; the other three are +// staged (below). // // The instruction and criteria of needs_human_decision are carried VERBATIM // from the day-73 eval (eval2.py, section D), which measured it at AUC 0.939 @@ -26,6 +28,15 @@ import ( // answers stay in item_decisions and read as stale questions, since no // registered question carries their keys. // +// THREE MORE QUESTIONS are STAGED here (TASK-3921, IDEA-3906 half B4a): +// effort, effort_medium and high_stakes. They ride the same call as +// needs_human_decision, so the cost is output tokens, not a second request. +// Nothing READS them yet: their reader is TASK-3959 (B4b), so the set lists +// them in Unsurfaced and no flag, chip or brief entry is made from them. Their +// answers are stored, and the item's decisions list returns them. The read +// rule that belongs to their reader, not to the set: an effort_medium answer +// whose top probability is under .5 means "unclear". +// // The eval's state was {title, body}; the runner sends the richer // BuildItemState (collection, fields, and the recent trail as well), so the // eval figure is evidence about the question, not a measurement of this @@ -35,6 +46,11 @@ const ( AttentionNeedsHuman = "needs_human_decision" + // Staged before their reader (TASK-3959); see the header. + AttentionEffort = "effort" + AttentionEffortMedium = "effort_medium" + AttentionHighStakes = "high_stakes" + // AttentionThreshold is the Noul probability at or above which the // dashboard surfaces an item. 0.7 sits between the eval's 0.5 operating // point (precision 93%, recall 76%) and the docs' 0.9 destructive-action @@ -52,6 +68,34 @@ func AttentionSet() QuestionSet { "the next step requires a human's decision, approval, sign-off, credentials, money, an external relationship, or hardware the agent cannot reach", "the next step is implementable by an engineer or AI agent from the description alone", ), + AttentionEffort: Score( + "How much effort does it take the human owner to supply what this work item is waiting on, or to do its next step?", + []string{ + "a yes/no or one-line answer", + "a short reply, or minutes at a computer", + "up to an hour", + "research, or hours of work", + }, + ), + AttentionEffortMedium: Choice( + "Where does the human owner have to be to supply what this work item is waiting on?", + map[string]string{ + "desk": "at a computer: writing, reading, a review or a setting", + "device": "at a particular device, account or piece of hardware, such as a phone, a test device or a console", + "other_person": "it needs another person's answer or action", + "unclear": "the text does not say", + }, + ), + AttentionHighStakes: Noul( + "Getting this work item wrong would be costly: it touches money, security, credentials, legal matters or a release date.", + "the work involves payments, secrets or access, a legal or compliance obligation, or a date someone is counting on", + "an ordinary change that is easy to correct afterwards", + ), + }, + Unsurfaced: map[string]bool{ + AttentionEffort: true, + AttentionEffortMedium: true, + AttentionHighStakes: true, }, Eligible: attentionEligible, Threshold: AttentionThreshold, @@ -171,6 +215,10 @@ func (r *Runner) WorkspaceFlags(workspaceID, setName string, threshold float64) byItem := make(map[string][]candidate) var order []string for _, d := range rows { + // A staged key (TASK-3921) makes no flag and no state read. + if qs.Unsurfaced[d.QuestionKey] { + continue + } p, v := surfacing(d, qhashNow[d.QuestionKey], threshold) if v != Surfaced { continue diff --git a/internal/decision/attention_test.go b/internal/decision/attention_test.go index 5f97663d0..71f4e3788 100644 --- a/internal/decision/attention_test.go +++ b/internal/decision/attention_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "net/http" + "reflect" "sync" "testing" "time" @@ -26,16 +27,45 @@ func TestAttentionSet_QuestionsAreTheEvalsVerbatim(t *testing.T) { TrueDesc: "the next step requires a human's decision, approval, sign-off, credentials, money, an external relationship, or hardware the agent cannot reach", FalseDesc: "the next step is implementable by an engineer or AI agent from the description alone", }, + // The three staged questions (TASK-3921), also pinned as literals. + "effort": { + Kind: KindScore, + Instructions: "How much effort does it take the human owner to supply what this work item is waiting on, or to do its next step?", + Levels: []string{ + "a yes/no or one-line answer", + "a short reply, or minutes at a computer", + "up to an hour", + "research, or hours of work", + }, + }, + "effort_medium": { + Kind: KindChoice, + Instructions: "Where does the human owner have to be to supply what this work item is waiting on?", + Options: map[string]string{ + "desk": "at a computer: writing, reading, a review or a setting", + "device": "at a particular device, account or piece of hardware, such as a phone, a test device or a console", + "other_person": "it needs another person's answer or action", + "unclear": "the text does not say", + }, + }, + "high_stakes": { + Kind: KindNoul, + Instructions: "Getting this work item wrong would be costly: it touches money, security, credentials, legal matters or a release date.", + TrueDesc: "the work involves payments, secrets or access, a legal or compliance obligation, or a date someone is counting on", + FalseDesc: "an ordinary change that is easy to correct afterwards", + }, } - got := AttentionSet().Questions + set := AttentionSet() + got := set.Questions for key, w := range want { g, ok := got[key] if !ok { t.Errorf("question %q missing from the attention set", key) continue } - if g.Kind != w.Kind || g.Instructions != w.Instructions || g.TrueDesc != w.TrueDesc || g.FalseDesc != w.FalseDesc { - t.Errorf("question %q differs from the eval's text:\n got %+v\nwant %+v", key, g, w) + if g.Kind != w.Kind || g.Instructions != w.Instructions || g.TrueDesc != w.TrueDesc || g.FalseDesc != w.FalseDesc || + !reflect.DeepEqual(g.Levels, w.Levels) || !reflect.DeepEqual(g.Options, w.Options) { + t.Errorf("question %q differs from its pinned text:\n got %+v\nwant %+v", key, g, w) } } // TASK-3915 / TASK-3924 retired the other two: the graph owns blocked, @@ -45,8 +75,15 @@ func TestAttentionSet_QuestionsAreTheEvalsVerbatim(t *testing.T) { t.Errorf("retired question %q is still asked", retired) } } - if len(got) != 1 { - t.Errorf("attention set has %d questions, want 1", len(got)) + if len(got) != 4 { + t.Errorf("attention set has %d questions, want 4", len(got)) + } + // The three staged keys, and only they, are unsurfaced. + if len(set.Unsurfaced) != 3 || !set.Unsurfaced["effort"] || !set.Unsurfaced["effort_medium"] || !set.Unsurfaced["high_stakes"] || set.Unsurfaced["needs_human_decision"] { + t.Errorf("Unsurfaced = %v, want exactly effort, effort_medium, high_stakes", set.Unsurfaced) + } + if set.MaxPerCall != 0 { + t.Errorf("MaxPerCall = %d; the set must go in one call", set.MaxPerCall) } } @@ -105,8 +142,19 @@ func newAttentionFixture(t *testing.T) *attentionFixture { fx.mu.Lock() defer fx.mu.Unlock() out := map[string]any{} - for key := range req.Questions { - out[key] = map[string]any{"type": "noul", "noul": fx.answers[key]} + for key, q := range req.Questions { + // Answer each question as ITS kind (the set mixes all three since + // TASK-3921); the provider refuses a mismatched answer. + switch q.Type { + case string(KindScore): + out[key] = map[string]any{"type": "score", "score": 1, "confidence": 0.9, + "probabilities": map[string]float64{"1": 1}} + case string(KindChoice): + out[key] = map[string]any{"type": "choice", "choice": "desk", "confidence": 0.9, + "probabilities": map[string]float64{"desk": 1}} + default: + out[key] = map[string]any{"type": "noul", "noul": fx.answers[key]} + } } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ @@ -125,6 +173,7 @@ func newAttentionFixture(t *testing.T) *attentionFixture { // tick; the re-ask debounce (TASK-3916) has its own tests, which this // would otherwise make every write-then-tick here wait an hour for. fx.s.SetDecisionDebounce(nil) + fx.r.backfillPoll = time.Millisecond // the backfill's drain wait polls fast return fx } @@ -337,8 +386,15 @@ func TestWorkspaceFlags_PendingAnswerStillSurfaced(t *testing.T) { if j, _ := fx.s.GetDecisionJob(it.ID, AttentionSetName); j == nil || j.LastError != "" { t.Fatalf("precondition: the comment owes a clean job, got %+v", j) } - if ds, _ := fx.r.Decisions(it.ID); len(ds) != 1 || ds[0].Current { - t.Fatalf("precondition: the item page must read the moved answer as NOT current, got %+v", ds) + // The item page returns every key's row (four since TASK-3921), none current. + ds, _ := fx.r.Decisions(it.ID) + if len(ds) != len(AttentionSet().Questions) { + t.Fatalf("precondition: want a row per attention question, got %d", len(ds)) + } + for _, d := range ds { + if d.Current { + t.Fatalf("precondition: the item page must read the moved answer as NOT current, got %+v", d) + } } flags, _, err = fx.r.WorkspaceFlags(fx.ws.ID, AttentionSetName, AttentionThreshold) if err != nil { @@ -387,8 +443,8 @@ func TestWorkspaceFlags_PendingAnswerStillSurfaced(t *testing.T) { } // Rewording ONE question of a set drops that key's answer and keeps the -// others'. The production attention set has one question since TASK-3915, -// so the set here adds a second one to keep the per-key rule exercised. +// others'. The production attention set has several questions, so the +// set here adds one more Noul that this test rewords. func TestWorkspaceFlags_RewordingOneQuestionKeepsTheOthers(t *testing.T) { fx := newAttentionFixture(t) two := AttentionSet() diff --git a/internal/decision/effort_backfill.go b/internal/decision/effort_backfill.go new file mode 100644 index 000000000..458e6a62f --- /dev/null +++ b/internal/decision/effort_backfill.go @@ -0,0 +1,208 @@ +package decision + +import ( + "context" + "fmt" + "log/slog" + "time" +) + +// The one-shot effort backfill (TASK-3921, IDEA-3906 B4a). +// +// effort, effort_medium and high_stakes joined the attention set after the +// set had already answered needs_human_decision for every item it covers. Left +// alone, an item would only learn its effort answers on its next write, so the +// staged questions would sit empty for most of the board when their reader +// (TASK-3959) arrives. The backfill owes ONE job per item that has a +// needs_human_decision answer and no effort answer, and lets the tick answer +// them under the re-ask debounce (an item evaluated inside the window waits out +// the rest of it). It makes NO provider call itself. +// +// It is gated by a platform_settings marker written only after every chunk has +// been owed, so an interrupted run owes the rest on the next one. Re-owing an +// item that already has a job coalesces into the same row (a generation bump), +// so a repeat is idempotent rather than additive. + +// EffortBackfillSetting is the platform_settings marker: set means every +// candidate was owed. +const EffortBackfillSetting = "effort_decisions_backfilled" + +// effortBackfillChunk is how many items one OweDecisionJobs call covers: one +// tick pass (the server's defaultDecisionTickLimit, which wires the live value +// through SetBackfillPacing). +const effortBackfillChunk = 20 + +// effortBackfillPoll is how often the backfill re-checks that its last chunk +// drained: the tick interval (the server wires the live value). +const effortBackfillPoll = 5 * time.Second + +// EffortBackfillCandidates returns the items the backfill would owe a job +// for, grouped by workspace in workspace order: live, answered for +// needs_human_decision, never answered for effort (any state or hash), and +// eligible for the attention set as it applies NOW (the runner's own rule, +// appliesNow). It owes nothing and calls no provider; it is the dry run. +func (r *Runner) EffortBackfillCandidates(ctx context.Context) (map[string][]string, []string, error) { + if r == nil { + return nil, nil, nil + } + qs, ok := r.registry.Get(AttentionSetName) + if !ok { + return nil, nil, ErrUnknownSet + } + cands, err := r.store.ItemsMissingDecisionKey(AttentionSetName, AttentionNeedsHuman, AttentionEffort) + if err != nil { + return nil, nil, err + } + byWS := map[string][]string{} + var order []string + for _, c := range cands { + if err := ctx.Err(); err != nil { + return nil, nil, err + } + item, err := r.store.GetItem(c.ItemID) + if err != nil { + return nil, nil, fmt.Errorf("decision: read item %s: %w", c.ItemID, err) + } + if item == nil || item.DeletedAt != nil { + continue + } + applies, err := r.appliesNow(qs, item) + if err != nil { + return nil, nil, err + } + if !applies { + continue + } + if _, seen := byWS[c.WorkspaceID]; !seen { + order = append(order, c.WorkspaceID) + } + byWS[c.WorkspaceID] = append(byWS[c.WorkspaceID], c.ItemID) + } + return byWS, order, nil +} + +// CountEffortBackfill is the dry run: how many jobs BackfillEffort would owe +// now. It ignores the marker and writes nothing. +func (r *Runner) CountEffortBackfill(ctx context.Context) (int, error) { + byWS, _, err := r.EffortBackfillCandidates(ctx) + if err != nil { + return 0, err + } + n := 0 + for _, ids := range byWS { + n += len(ids) + } + return n, nil +} + +// BackfillEffort owes the attention set to every candidate, then writes the +// marker. owed is the number of jobs owed by this call (0 when the marker was +// already set). A nil runner does nothing. +// +// PACING (lead ruling on TASK-3921, r3 M1). Claims run oldest-enqueued first +// and evaluation is sequential, so owing every candidate at once would put a +// live edit's re-ask behind the whole backlog. The backfill therefore owes one +// tick pass of items (backfillChunk) and owes the next only once THAT chunk's +// own jobs have drained, where drained means none of the chunk's items has a +// job ClaimDecisionJobs could take right now (CountClaimableDecisionJobs: the +// same predicate). It does NOT gate on total queue depth, which would starve +// the backfill under live load. A job held by the re-ask debounce (not_before +// ahead, up to ~1h) cannot be claimed yet and must not stall the run; a +// claimed, answered or deleted job is drained. Live jobs owed meanwhile queue +// behind at most one chunk. The wait polls at the tick interval and honours +// ctx; a cancel in it returns ctx.Err() with no marker. The marker is written +// after the LAST chunk is owed, without waiting for it to drain. +// +// ctx is checked before each candidate read and before each chunk. A cancel +// that lands before a chunk returns ctx.Err() with no marker, so the next run +// redoes the work. A cancel that lands after the last chunk may still write +// the marker, which is correct: the marker means every candidate was ENQUEUED, +// not answered (lead ruling on TASK-3921, mutant M5). +func (r *Runner) BackfillEffort(ctx context.Context) (owed int, err error) { + if r == nil { + return 0, nil + } + done, err := r.store.GetPlatformSetting(EffortBackfillSetting) + if err != nil { + return 0, fmt.Errorf("decision: read %s: %w", EffortBackfillSetting, err) + } + if done != "" { + return 0, nil + } + byWS, order, err := r.EffortBackfillCandidates(ctx) + if err != nil { + return 0, err + } + chunk := r.backfillChunk + if chunk <= 0 { + chunk = effortBackfillChunk + } + n := 0 + var prev []string // the previous chunk's items, awaiting drain + for _, ws := range order { + ids := byWS[ws] + for start := 0; start < len(ids); start += chunk { + end := start + chunk + if end > len(ids) { + end = len(ids) + } + if prev != nil { + if err := r.waitChunkDrained(ctx, prev); err != nil { + return owed, err + } + } + if r.beforeBackfillChunk != nil { + if err := r.beforeBackfillChunk(ctx, n); err != nil { + return owed, err + } + } + // The chunk-boundary cancel check (single line on purpose). + if err := ctx.Err(); err != nil { + return owed, err + } + if err := r.store.OweDecisionJobs(ws, AttentionSetName, ids[start:end]); err != nil { + return owed, err + } + owed += end - start + prev = ids[start:end] + n++ + } + } + // ONLY NOW: every chunk was owed. + if err := r.store.SetPlatformSetting(EffortBackfillSetting, "1"); err != nil { + return owed, fmt.Errorf("decision: persist %s: %w", EffortBackfillSetting, err) + } + l := r.logger + if l == nil { + l = slog.Default() + } + l.Info("effort backfill: owed attention jobs for items with no effort answer", "owed", owed) + return owed, nil +} + +// waitChunkDrained blocks until none of ids has a job claimable now (see +// BackfillEffort), polling at backfillPoll. The check runs first, so an +// already-drained chunk costs one query. +func (r *Runner) waitChunkDrained(ctx context.Context, ids []string) error { + poll := r.backfillPoll + if poll <= 0 { + poll = effortBackfillPoll + } + for { + if err := ctx.Err(); err != nil { + return err + } + claimable, err := r.store.CountClaimableDecisionJobs(AttentionSetName, ids, time.Now()) + if err != nil { + return err + } + if claimable == 0 { // THE DRAIN-WAIT LINE: mutate to skip the wait + return nil + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(poll): + } + } +} diff --git a/internal/decision/effort_backfill_test.go b/internal/decision/effort_backfill_test.go new file mode 100644 index 000000000..0f20d2c9f --- /dev/null +++ b/internal/decision/effort_backfill_test.go @@ -0,0 +1,659 @@ +package decision + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/models" + "github.com/PerpetualSoftware/pad/internal/store" +) + +// TASK-3921: the staged attention questions and their one-shot backfill. + +const evalTimeFormat = "2006-01-02T15:04:05.000000000Z" + +// seedAnswers stores a Noul-shaped row per key for the item in the attention +// set, evaluated at `at`, as the runner would have left them. +func (fx *attentionFixture) seedAnswers(t *testing.T, it *models.Item, at time.Time, keys ...string) { + t.Helper() + rows := make([]models.ItemDecision, 0, len(keys)) + for _, k := range keys { + rows = append(rows, models.ItemDecision{ + ItemID: it.ID, QuestionSet: AttentionSetName, QuestionKey: k, Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.9}`), Provider: "fake", Model: "m", + StateHash: "h", QuestionHash: "q-" + k, ItemSeq: 1, EvaluatedAt: at.UTC().Format(evalTimeFormat), + }) + } + if err := fx.s.InsertItemDecisions(fx.ws.ID, rows); err != nil { + t.Fatal(err) + } +} + +// clearJobs drops the jobs the item creations owed, so a test sees only what +// the backfill owes. +func (fx *attentionFixture) clearJobs(t *testing.T) { + t.Helper() + if _, err := fx.s.DB().Exec(`DELETE FROM decision_jobs`); err != nil { + t.Fatal(err) + } +} + +// jobItems returns item_id -> number of rows, for the attention set. +func (fx *attentionFixture) jobItems(t *testing.T) map[string]int { + t.Helper() + rows, err := fx.s.DB().Query(`SELECT item_id, question_set FROM decision_jobs`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + out := map[string]int{} + for rows.Next() { + var id, set string + if err := rows.Scan(&id, &set); err != nil { + t.Fatal(err) + } + if set != AttentionSetName { + t.Fatalf("the backfill owed a job for set %q; it must owe the attention set only", set) + } + out[id]++ + } + return out +} + +func (fx *attentionFixture) marker(t *testing.T) string { + t.Helper() + v, err := fx.s.GetPlatformSetting(EffortBackfillSetting) + if err != nil { + t.Fatal(err) + } + return v +} + +// holdDrained plays a tick that claims every job as it appears: it marks each +// decision_jobs row claimed under a far lease (rows stay, so tests still count +// them) until the test ends. A multi-chunk backfill needs it, since each chunk +// waits for the previous one's jobs to be taken. +func (fx *attentionFixture) holdDrained(t *testing.T) { + t.Helper() + stop := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + far := time.Now().Add(time.Hour).UTC().Format(evalTimeFormat) + for { + select { + case <-stop: + return + case <-time.After(time.Millisecond): + _, _ = fx.s.DB().Exec(`UPDATE decision_jobs SET claimed_by = 'test', lease_expires_at = ? WHERE claimed_by IS NULL`, far) + } + } + }() + t.Cleanup(func() { close(stop); <-done }) +} + +// seedCandidates makes n open items answered for needs_human_decision only and +// clears the jobs their creation owed. +func (fx *attentionFixture) seedCandidates(t *testing.T, n int) []string { + t.Helper() + var ids []string + for i := 0; i < n; i++ { + it := fx.item(t, fx.tasks.ID, "open") + fx.seedAnswers(t, it, time.Now().Add(-3*time.Hour), AttentionNeedsHuman) + ids = append(ids, it.ID) + } + fx.clearJobs(t) + return ids +} + +// waitJobs polls until at least n decision_jobs rows exist. +func (fx *attentionFixture) waitJobs(t *testing.T, n int) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for { + var c int + if err := fx.s.DB().QueryRow(`SELECT COUNT(*) FROM decision_jobs`).Scan(&c); err != nil { + t.Fatal(err) + } + if c >= n { + return + } + if time.Now().After(deadline) { + t.Fatalf("never saw %d decision_jobs rows", n) + } + time.Sleep(time.Millisecond) + } +} + +var allAttentionKeys = []string{AttentionNeedsHuman, AttentionEffort, AttentionEffortMedium, AttentionHighStakes} + +func TestEffortBackfill_RunsOnceAndSkipsUnasked(t *testing.T) { + fx := newAttentionFixture(t) + now := time.Now().Add(-3 * time.Hour) + sys, err := fx.s.CreateCollection(fx.ws.ID, models.CollectionCreate{Name: "Conventions", Schema: statusSchema, IsSystem: true}) + if err != nil { + t.Fatal(err) + } + a := fx.item(t, fx.tasks.ID, "open") // needs_human only: owed + b := fx.item(t, fx.tasks.ID, "open") // already has effort: skipped + c := fx.item(t, fx.tasks.ID, "open") // never answered: skipped + d := fx.item(t, fx.tasks.ID, "done") // terminal: skipped + e := fx.item(t, sys.ID, "open") // system collection: skipped + f := fx.item(t, fx.tasks.ID, "open") // soft-deleted: skipped + fx.seedAnswers(t, a, now, AttentionNeedsHuman) + fx.seedAnswers(t, b, now, allAttentionKeys...) + fx.seedAnswers(t, d, now, AttentionNeedsHuman) + fx.seedAnswers(t, e, now, AttentionNeedsHuman) + fx.seedAnswers(t, f, now, AttentionNeedsHuman) + _ = c + if err := fx.s.DeleteItem(f.ID); err != nil { + t.Fatal(err) + } + fx.clearJobs(t) + + // The dry run counts and writes nothing. + n, err := fx.r.CountEffortBackfill(context.Background()) + if err != nil || n != 1 { + t.Fatalf("dry run = %d, %v; want 1", n, err) + } + if got := fx.jobItems(t); len(got) != 0 || fx.marker(t) != "" { + t.Fatalf("the dry run wrote: jobs %v marker %q", got, fx.marker(t)) + } + + owed, err := fx.r.BackfillEffort(context.Background()) + if err != nil || owed != 1 { + t.Fatalf("BackfillEffort = %d, %v; want 1", owed, err) + } + if got := fx.jobItems(t); len(got) != 1 || got[a.ID] != 1 { + t.Fatalf("jobs = %v; want only item a", got) + } + if fx.marker(t) == "" { + t.Fatal("the marker was not written") + } + + // Second run: the marker stops it, even with a's job gone and a still a + // candidate (it has no effort answer yet). + fx.clearJobs(t) + owed, err = fx.r.BackfillEffort(context.Background()) + if err != nil || owed != 0 { + t.Fatalf("second run = %d, %v; want 0", owed, err) + } + if got := fx.jobItems(t); len(got) != 0 { + t.Fatalf("second run owed %v", got) + } +} + +func TestEffortBackfill_FailureLeavesNoMarker(t *testing.T) { + fx := newAttentionFixture(t) + a := fx.item(t, fx.tasks.ID, "open") + fx.seedAnswers(t, a, time.Now().Add(-3*time.Hour), AttentionNeedsHuman) + fx.clearJobs(t) + boom := errors.New("boom") + fx.r.beforeBackfillChunk = func(context.Context, int) error { return boom } + if _, err := fx.r.BackfillEffort(context.Background()); !errors.Is(err, boom) { + t.Fatalf("err = %v; want the injected failure", err) + } + if fx.marker(t) != "" { + t.Fatal("a failed backfill wrote the marker") + } +} + +// A crash between chunks: the second chunk fails, no marker; a re-run owes +// every candidate, one row each, and writes the marker. +func TestEffortBackfill_FailingSecondChunkThenRerun(t *testing.T) { + fx := newAttentionFixture(t) + var ids []string + for i := 0; i < 3; i++ { + it := fx.item(t, fx.tasks.ID, "open") + fx.seedAnswers(t, it, time.Now().Add(-3*time.Hour), AttentionNeedsHuman) + ids = append(ids, it.ID) + } + fx.clearJobs(t) + fx.r.backfillChunk = 1 + fx.holdDrained(t) // chunk k+1 waits for chunk k's jobs to be taken + boom := errors.New("crash") + fx.r.beforeBackfillChunk = func(_ context.Context, chunk int) error { + if chunk == 1 { + return boom + } + return nil + } + if _, err := fx.r.BackfillEffort(context.Background()); !errors.Is(err, boom) { + t.Fatalf("err = %v; want the injected failure", err) + } + if fx.marker(t) != "" { + t.Fatal("marker written after a failed second chunk") + } + if got := fx.jobItems(t); len(got) != 1 { + t.Fatalf("after the crash jobs = %v; want the first chunk's one", got) + } + + fx.r.beforeBackfillChunk = nil + owed, err := fx.r.BackfillEffort(context.Background()) + if err != nil || owed != 3 { + t.Fatalf("re-run = %d, %v; want 3", owed, err) + } + got := fx.jobItems(t) + sort.Strings(ids) + if len(got) != 3 { + t.Fatalf("jobs = %v; want one per candidate", got) + } + for _, id := range ids { + if got[id] != 1 { + t.Fatalf("item %s has %d job rows; want exactly 1", id, got[id]) + } + } + if fx.marker(t) == "" { + t.Fatal("marker not written after the completed re-run") + } +} + +// The backfill's jobs go through the installed debounce: an item evaluated +// 10 minutes ago waits out the hour, one evaluated 2 hours ago does not. +func TestEffortBackfill_RespectsDebounce(t *testing.T) { + fx := newAttentionFixture(t) + fx.r.Install(fx.s) // the fixture turned the debounce off + recent := fx.item(t, fx.tasks.ID, "open") + old := fx.item(t, fx.tasks.ID, "open") + fx.seedAnswers(t, recent, time.Now().Add(-10*time.Minute), AttentionNeedsHuman) + fx.seedAnswers(t, old, time.Now().Add(-2*time.Hour), AttentionNeedsHuman) + fx.clearJobs(t) + if owed, err := fx.r.BackfillEffort(context.Background()); err != nil || owed != 2 { + t.Fatalf("BackfillEffort = %d, %v", owed, err) + } + jr, err := fx.s.GetDecisionJob(recent.ID, AttentionSetName) + if err != nil || jr == nil { + t.Fatalf("recent job: %+v %v", jr, err) + } + if jr.NotBefore == "" || jr.NotBefore <= time.Now().UTC().Format(evalTimeFormat) { + t.Fatalf("recent job not_before = %q; want a future time", jr.NotBefore) + } + jo, err := fx.s.GetDecisionJob(old.ID, AttentionSetName) + if err != nil || jo == nil { + t.Fatalf("old job: %+v %v", jo, err) + } + if jo.NotBefore != "" { + t.Fatalf("old job not_before = %q; want none", jo.NotBefore) + } +} + +func TestEffortBackfill_NoCallsOnlyJobs(t *testing.T) { + fx := newAttentionFixture(t) + a := fx.item(t, fx.tasks.ID, "open") + fx.seedAnswers(t, a, time.Now().Add(-3*time.Hour), AttentionNeedsHuman) + fx.clearJobs(t) + before := len(fx.f.requests) + if _, err := fx.r.BackfillEffort(context.Background()); err != nil { + t.Fatal(err) + } + if got := len(fx.f.requests) - before; got != 0 { + t.Fatalf("the backfill made %d provider calls; want 0", got) + } + if len(fx.jobItems(t)) != 1 { + t.Fatal("precondition: the backfill owed no job") + } + // And the tick then answers the owed job with ONE call carrying all four. + fx.tick(t) + if len(fx.f.requests) != before+1 || len(fx.f.requests[before].Questions) != 4 { + t.Fatalf("tick requests = %d (questions %v)", len(fx.f.requests)-before, fx.f.requests) + } +} + +// Readers: one test per reader that consults decisions, each of which fails +// when that reader's Unsurfaced line is dropped. + +// stagedFixture leaves an item whose only above-threshold answer is the staged +// high_stakes (0.95); needs_human_decision is below threshold. +func stagedFixture(t *testing.T, needsHuman float64) (*attentionFixture, *models.Item) { + t.Helper() + fx := newAttentionFixture(t) + fx.set(AttentionNeedsHuman, needsHuman) + fx.set(AttentionHighStakes, 0.95) + it := fx.item(t, fx.tasks.ID, "open") + fx.tick(t) + if ds, err := fx.r.Decisions(it.ID); err != nil || len(ds) != 4 { + t.Fatalf("precondition: Decisions returned %d rows, %v; want all four", len(ds), err) + } + return fx, it +} + +func TestWorkspaceFlags_IgnoresStagedKeys(t *testing.T) { + fx, it := stagedFixture(t, 0.3) + flags, _, err := fx.r.WorkspaceFlags(fx.ws.ID, AttentionSetName, AttentionThreshold) + if err != nil { + t.Fatal(err) + } + if got, ok := flags[it.ID]; ok { + t.Fatalf("high_stakes at .95 flagged the item: %v", got) + } +} + +func TestSurfacedDecisions_HidesStagedKeys(t *testing.T) { + fx, it := stagedFixture(t, 0.9) + got, _, _, err := fx.r.SurfacedDecisions(it.ID) + if err != nil { + t.Fatal(err) + } + if keys := surfacedKeys(got); len(keys) != 1 || keys[0] != AttentionNeedsHuman { + t.Fatalf("surfaced = %v; want only needs_human_decision", keys) + } +} + +// The staged keys are not counted anywhere: the score and choice rows would +// each read as below threshold, and the low high_stakes with them. +func TestSurfacedDecisions_StagedKeysAreNotCounted(t *testing.T) { + fx := newAttentionFixture(t) + fx.set(AttentionNeedsHuman, 0.9) + fx.set(AttentionHighStakes, 0.1) + it := fx.item(t, fx.tasks.ID, "open") + fx.tick(t) + _, below, stale, err := fx.r.SurfacedDecisions(it.ID) + if err != nil { + t.Fatal(err) + } + if below != 0 || stale != 0 { + t.Fatalf("below=%d stale=%d; the staged keys must count in neither", below, stale) + } +} + +// Dry run against a database FILE (a copy of the live one), for measuring the +// backfill before it ships. It calls no provider and writes nothing but the +// migrations store.New applies to the file, so run it on a COPY: +// +// PAD_EFFORT_DRYRUN_DB=/path/to/copy.db go test ./internal/decision/ -run TestEffortBackfillDryRunOnFile -v -count=1 +func TestEffortBackfillDryRunOnFile(t *testing.T) { + path := os.Getenv("PAD_EFFORT_DRYRUN_DB") + if path == "" { + t.Skip("set PAD_EFFORT_DRYRUN_DB to a COPY of a pad database") + } + if err := refuseLiveDatabase(path, os.Getenv("HOME")); err != nil { + t.Fatalf("%v", err) + } + // Open a private COPY, never the named file: store.New migrates what it + // opens. The copy lives under GOTMPDIR (disk), not /tmp (RAM tmpfs). + dir := t.TempDir() + if g := os.Getenv("GOTMPDIR"); g != "" { + d, err := os.MkdirTemp(g, "effort-dryrun-") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(d) }) + dir = d + } + copyPath := filepath.Join(dir, "copy.db") + src, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(copyPath, src, 0o600); err != nil { + t.Fatal(err) + } + s, err := store.New(copyPath) + if err != nil { + t.Fatalf("open %s: %v", copyPath, err) + } + defer s.Close() + reg, err := ProductionRegistry() + if err != nil { + t.Fatal(err) + } + // The provider is never called; the runner only needs one to exist. + r := NewRunner(s, newTypesafe("unused", "unused"), reg) + byWS, order, err := r.EffortBackfillCandidates(context.Background()) + if err != nil { + t.Fatal(err) + } + total := 0 + for _, ws := range order { + t.Logf("workspace %s: %d items", ws, len(byWS[ws])) + total += len(byWS[ws]) + } + t.Logf("EFFORT BACKFILL WOULD OWE %d JOBS", total) +} + +// refuseLiveDatabase guards the dry run: store.New MIGRATES the file it opens, +// so a live database must never be pointed at. A -wal or -shm sibling means it +// is open (or was not closed cleanly), and anything under $HOME/.pad is the +// real data directory. +func refuseLiveDatabase(path, home string) error { + abs, err := filepath.Abs(path) + if err != nil { + return err + } + recipe := fmt.Sprintf("take a copy first: sqlite3 'file:%s?mode=ro' \".backup ''\"", abs) + for _, suffix := range []string{"-wal", "-shm"} { + if _, err := os.Stat(abs + suffix); err == nil { + return fmt.Errorf("refusing %s: %s%s exists, so it looks live; %s", abs, abs, suffix, recipe) + } + } + if home != "" { + padDir := filepath.Join(filepath.Clean(home), ".pad") + string(filepath.Separator) + if strings.HasPrefix(abs, padDir) { + return fmt.Errorf("refusing %s: it is under %s; %s", abs, padDir, recipe) + } + } + return nil +} + +func TestRefuseLiveDatabase(t *testing.T) { + dir := t.TempDir() + db := filepath.Join(dir, "copy.db") + if err := os.WriteFile(db, nil, 0o600); err != nil { + t.Fatal(err) + } + if err := refuseLiveDatabase(db, dir+"-other"); err != nil { + t.Fatalf("a clean copy was refused: %v", err) + } + for _, suffix := range []string{"-wal", "-shm"} { + sib := db + suffix + if err := os.WriteFile(sib, nil, 0o600); err != nil { + t.Fatal(err) + } + if err := refuseLiveDatabase(db, dir+"-other"); err == nil || !strings.Contains(err.Error(), ".backup") { + t.Errorf("a %s sibling was not refused with the recipe: %v", suffix, err) + } + os.Remove(sib) + } + home := t.TempDir() + padDB := filepath.Join(home, ".pad", "pad.db") + if err := refuseLiveDatabase(padDB, home); err == nil { + t.Error("a path under $HOME/.pad was not refused") + } +} + +// A cancel between chunks stops the run with ctx.Err(), owes nothing more and +// never writes the marker; a pre-cancelled context reads no candidate either. +func TestEffortBackfill_CancelStopsBetweenChunksAndWritesNoMarker(t *testing.T) { + fx := newAttentionFixture(t) + for i := 0; i < 3; i++ { + it := fx.item(t, fx.tasks.ID, "open") + fx.seedAnswers(t, it, time.Now().Add(-3*time.Hour), AttentionNeedsHuman) + } + fx.clearJobs(t) + + dead, cancelDead := context.WithCancel(context.Background()) + cancelDead() + if _, _, err := fx.r.EffortBackfillCandidates(dead); !errors.Is(err, context.Canceled) { + t.Fatalf("candidate read on a cancelled ctx: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + fx.r.backfillChunk = 1 + fx.holdDrained(t) // chunk k+1 waits for chunk k's jobs to be taken + fx.r.beforeBackfillChunk = func(_ context.Context, chunk int) error { + if chunk == 1 { + cancel() // lands inside the hook; the chunk-boundary check must see it + } + return nil + } + owed, err := fx.r.BackfillEffort(ctx) + if !errors.Is(err, context.Canceled) || owed != 1 { + t.Fatalf("owed=%d err=%v; want 1 and context.Canceled", owed, err) + } + if got := fx.jobItems(t); len(got) != 1 { + t.Fatalf("jobs after cancel = %v; want only the first chunk's", got) + } + if fx.marker(t) != "" { + t.Fatal("marker written by a cancelled run") + } +} + +// A chunk whose WRITE fails in the store, not in the test hook, must stop the +// run before the marker. The hook only breaks the table: the error has to come +// back through OweDecisionJobs, which is the path the other failure tests +// never reach (they fail the hook itself). It breaks the table before the LAST +// chunk on purpose: after any earlier chunk, the drain wait that follows reads +// the same table and fails too, which would keep the marker unwritten even if +// the chunk's own error were ignored (mutant M6, TASK-3921). +func TestEffortBackfill_StoreFailureMidRunWritesNoMarker(t *testing.T) { + fx := newAttentionFixture(t) + for i := 0; i < 3; i++ { + it := fx.item(t, fx.tasks.ID, "open") + fx.seedAnswers(t, it, time.Now().Add(-3*time.Hour), AttentionNeedsHuman) + } + fx.clearJobs(t) + fx.r.backfillChunk = 1 + fx.holdDrained(t) // chunk k+1 waits for chunk k's jobs to be taken + fx.r.beforeBackfillChunk = func(_ context.Context, chunk int) error { + if chunk == 2 { + if _, err := fx.s.DB().Exec(`ALTER TABLE decision_jobs RENAME TO decision_jobs_gone`); err != nil { + t.Fatal(err) + } + } + return nil + } + if _, err := fx.r.BackfillEffort(context.Background()); err == nil { + t.Fatal("a failed chunk write returned no error") + } + if fx.marker(t) != "" { + t.Fatal("marker written although a chunk's jobs were never owed") + } + if _, err := fx.s.DB().Exec(`ALTER TABLE decision_jobs_gone RENAME TO decision_jobs`); err != nil { + t.Fatal(err) + } + fx.r.beforeBackfillChunk = nil + if _, err := fx.r.BackfillEffort(context.Background()); err != nil { + t.Fatalf("re-run: %v", err) + } + if fx.marker(t) == "" { + t.Fatal("re-run after the store recovered wrote no marker") + } + if got := fx.jobItems(t); len(got) != 3 { + t.Fatalf("re-run owed %d items; want 3", len(got)) + } +} + +// Pacing (TASK-3921 r3 M1). The drain wait is ONE condition, marked +// "THE DRAIN-WAIT LINE" in waitChunkDrained (effort_backfill.go); making it +// return at once makes the next two tests go red. + +// A live job owed after chunk 1 is claimed before chunk 2's jobs are owed. +func TestEffortBackfill_LiveReaskInterleavesWithChunks(t *testing.T) { + fx := newAttentionFixture(t) + fx.seedCandidates(t, 3) + fx.r.backfillChunk = 1 + done := make(chan error, 1) + go func() { _, err := fx.r.BackfillEffort(context.Background()); done <- err }() + + fx.waitJobs(t, 1) // chunk 0 is owed and (nobody claims yet) waiting + // A normal write (a new item owes its job): it queues behind chunk 0. + live := fx.item(t, fx.tasks.ID, "open") + // Now a tick: claim one job per pass, oldest first, recording the order. + var order []string + deadline := time.Now().Add(10 * time.Second) + for len(order) < 4 { + if time.Now().After(deadline) { + t.Fatalf("claimed only %v", order) + } + js, err := fx.s.ClaimDecisionJobs("test", 1, time.Hour) + if err != nil { + t.Fatal(err) + } + for _, j := range js { + order = append(order, j.ItemID) + } + time.Sleep(time.Millisecond) + } + if err := <-done; err != nil { + t.Fatal(err) + } + if order[1] != live.ID { + t.Fatalf("claim order %v: the live re-ask (%s) must come right after chunk 0, before chunk 1's job", order, live.ID) + } +} + +// With chunk 0's jobs left claimable, chunk 1 is not owed; a cancel in the +// wait returns ctx.Err() with no chunk-1 job and no marker. +func TestEffortBackfill_HoldsNextChunkUntilDrainedThenCancelWritesNoMarker(t *testing.T) { + fx := newAttentionFixture(t) + fx.seedCandidates(t, 3) + fx.r.backfillChunk = 1 + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + type res struct { + owed int + err error + } + done := make(chan res, 1) + go func() { o, err := fx.r.BackfillEffort(ctx); done <- res{o, err} }() + fx.waitJobs(t, 1) + time.Sleep(100 * time.Millisecond) // ~100 polls at 1ms + if got := fx.jobItems(t); len(got) != 1 { + t.Fatalf("jobs while chunk 0 is undrained = %v; want only chunk 0's", got) + } + cancel() + r := <-done + if !errors.Is(r.err, context.Canceled) || r.owed != 1 { + t.Fatalf("owed=%d err=%v; want 1 and context.Canceled", r.owed, r.err) + } + if got := fx.jobItems(t); len(got) != 1 { + t.Fatalf("jobs after the cancel = %v", got) + } + if fx.marker(t) != "" { + t.Fatal("marker written by a run cancelled in the drain wait") + } +} + +// A chunk whose jobs are ALL debounced (not_before ahead) cannot be claimed and +// so must not hold the next chunk; the run completes and marks. +func TestEffortBackfill_DebouncedChunkDoesNotHoldTheNext(t *testing.T) { + fx := newAttentionFixture(t) + fx.seedCandidates(t, 3) + fx.r.backfillChunk = 1 + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + done := make(chan error, 1) + go func() { _, err := fx.r.BackfillEffort(ctx); done <- err }() + // Debounce every job as it appears (an item evaluated minutes ago). + future := time.Now().Add(time.Hour).UTC().Format(evalTimeFormat) + stop := make(chan struct{}) + go func() { + for { + select { + case <-stop: + return + case <-time.After(time.Millisecond): + _, _ = fx.s.DB().Exec(`UPDATE decision_jobs SET not_before = ? WHERE not_before IS NULL`, future) + } + } + }() + err := <-done + close(stop) + if err != nil { + t.Fatalf("a debounced chunk held the backfill: %v", err) + } + if got := fx.jobItems(t); len(got) != 3 { + t.Fatalf("jobs = %v; want all three owed", got) + } + if fx.marker(t) == "" { + t.Fatal("marker not written") + } +} diff --git a/internal/decision/runner.go b/internal/decision/runner.go index b4b602f04..6a51c24da 100644 --- a/internal/decision/runner.go +++ b/internal/decision/runner.go @@ -41,6 +41,14 @@ type QuestionSet struct { // with every answer, so renaming one orphans its history. Questions map[string]Question + // Unsurfaced names question keys that are asked and stored but that no + // surfacing read acts on yet (TASK-3921: staged before their reader). + // WorkspaceFlags, SurfacedDecisions (counters included) and + // ItemReadDecisions (item show, its --agent projection and MCP item get) + // skip them; Decisions, and the decisions endpoint that reads it, still + // return the rows. + Unsurfaced map[string]bool + // Eligible, when set, narrows the set to items whose CURRENT state it // accepts. It is checked by the runner at evaluation time, never by the // store's resolver: the resolver runs inside the write transaction and @@ -373,6 +381,13 @@ type Runner struct { // rollout). The server wires it to its metrics. usage func(set string, u Usage) + // backfillChunk overrides effortBackfillChunk when positive, and + // beforeBackfillChunk, when set, runs before each chunk is owed (chunk + // index from 0) and may fail the backfill there. Both are TEST-ONLY seams. + backfillChunk int + backfillPoll time.Duration + beforeBackfillChunk func(ctx context.Context, chunk int) error + // logger receives the per-evaluation spend line; nil means slog.Default(). // A field rather than the global so a test can read its own lines. logger *slog.Logger @@ -762,6 +777,49 @@ func (r *Runner) Decisions(itemID string) ([]models.ItemDecision, error) { return rows, err } +// SetBackfillSeams sets the effort backfill's chunk size and its +// before-each-chunk hook. Test affordance only (the server's tests live in +// another package). +func (r *Runner) SetBackfillSeams(chunk int, hook func(ctx context.Context, chunk int) error) { + if r == nil { + return + } + if chunk > 0 { + r.backfillChunk = chunk + } + r.beforeBackfillChunk = hook +} + +// SetBackfillPacing sets how many items one backfill chunk owes (the tick's +// per-pass limit) and how often the backfill re-checks that the chunk drained +// (the tick interval). Non-positive values keep the defaults. +func (r *Runner) SetBackfillPacing(chunk int, poll time.Duration) { + if r == nil { + return + } + r.backfillChunk = chunk + r.backfillPoll = poll +} + +// ItemReadDecisions is Decisions without the staged keys (TASK-3921): what the +// item read, the CLI --agent projection and MCP item get carry. The raw +// Decisions stays behind the decisions endpoint, which shows staged data. +func (r *Runner) ItemReadDecisions(itemID string) ([]models.ItemDecision, error) { + rows, err := r.Decisions(itemID) + if err != nil { + return nil, err + } + out := make([]models.ItemDecision, 0, len(rows)) + for _, d := range rows { + qs, _ := r.registry.Get(d.QuestionSet) + if qs.Unsurfaced[d.QuestionKey] { // the one filter line + continue + } + out = append(out, d) + } + return out, nil +} + // SurfacedDecisions is the item's latest answers that its set would surface // (TASK-3618, for the item brief): the question is current and the answer // reaches the set's Threshold, by the same row test WorkspaceFlags applies. @@ -781,6 +839,10 @@ func (r *Runner) SurfacedDecisions(itemID string) (surfaced []models.ItemDecisio surfaced = []models.ItemDecision{} for i, d := range rows { qs, _ := r.registry.Get(d.QuestionSet) + // A staged key (TASK-3921) is not surfaced and not counted. + if qs.Unsurfaced[d.QuestionKey] { + continue + } switch _, v := surfacing(d, qhashes[i], qs.Threshold); v { case Surfaced: surfaced = append(surfaced, d) diff --git a/internal/decision/task3618_surfaced_test.go b/internal/decision/task3618_surfaced_test.go index 4da11f1d5..3ffbca623 100644 --- a/internal/decision/task3618_surfaced_test.go +++ b/internal/decision/task3618_surfaced_test.go @@ -21,7 +21,7 @@ func surfacedKeys(ds []models.ItemDecision) []string { // TASK-3618: SurfacedDecisions keeps exactly the answers WorkspaceFlags would // flag, on both sides of the threshold (AT it is surfaced), and counts the // rest as below threshold. One item answers AT the threshold, the other just -// under it (the set has one question since TASK-3915). +// under it (needs_human_decision is the one surfaced question since TASK-3921; the other three are staged). func TestTASK3618_SurfacedDecisionsThresholdBothSides(t *testing.T) { fx := newAttentionFixture(t) fx.set(AttentionNeedsHuman, AttentionThreshold) diff --git a/internal/server/dashboard_attention_decisions_test.go b/internal/server/dashboard_attention_decisions_test.go index 4f88498dd..7a20aaa52 100644 --- a/internal/server/dashboard_attention_decisions_test.go +++ b/internal/server/dashboard_attention_decisions_test.go @@ -263,8 +263,9 @@ func TestDashboardAttention_ItemMadeTerminalBySchemaEditIsNotSurfaced(t *testing // answer is still stored, and none is current. (Before the edit they // were — the dashboard precondition above is that leg.) b := getDecisions(t, srv, slug, it.Slug) - if len(b.Decisions) != 1 { - t.Fatalf("decisions = %d rows, want the 1 stored answer", len(b.Decisions)) + // One row per attention question (four since TASK-3921), staged keys included. + if want := len(decision.AttentionSet().Questions); len(b.Decisions) != want { + t.Fatalf("decisions = %d rows, want the %d stored answers", len(b.Decisions), want) } for _, d := range b.Decisions { if d.Current { diff --git a/internal/server/decision_settings.go b/internal/server/decision_settings.go index a9fb68a02..348e6ca51 100644 --- a/internal/server/decision_settings.go +++ b/internal/server/decision_settings.go @@ -1,6 +1,7 @@ package server import ( + "context" "errors" "log/slog" "net/http" @@ -46,6 +47,11 @@ type decisionSettingsState struct { mu sync.Mutex file decision.Config sets *decision.Registry + + // backfillChunk and backfillHook are TEST-ONLY seams applied to every + // runner a reconfigure builds (see Runner.SetBackfillSeams). + backfillChunk int + backfillHook func(ctx context.Context, chunk int) error } // SetDecisionBase records the config-file source and the question-set @@ -79,6 +85,23 @@ func (s *Server) reconfigureDecisionsLocked() error { provider = nil } runner := decision.NewRunner(s.store, provider, sets) + if runner != nil { + // Pace the effort backfill by one tick pass (TASK-3921): the tick's + // own per-pass limit and interval, defaults when not yet started. + s.decisionTick.mu.Lock() + chunk, poll := s.decisionTick.limit, s.decisionTick.interval + s.decisionTick.mu.Unlock() + if chunk == 0 { + chunk = defaultDecisionTickLimit + } + if poll == 0 { + poll = defaultDecisionTickInterval + } + runner.SetBackfillPacing(chunk, poll) + } + if runner != nil && s.decisionSettings.backfillHook != nil { + runner.SetBackfillSeams(s.decisionSettings.backfillChunk, s.decisionSettings.backfillHook) + } if runner != nil && s.metrics != nil { m := s.metrics runner.SetUsageObserver(func(set string, u decision.Usage) { @@ -87,13 +110,72 @@ func (s *Server) reconfigureDecisionsLocked() error { m.DecisionTokensTotal.WithLabelValues(set, "output").Add(float64(u.OutputTokens)) }) } + // A running backfill belongs to the OLD runner and its provider: cancel it + // first, then start a new one only for a non-nil runner. + s.cancelEffortBackfill() s.SetDecisionRunner(runner) if runner != nil { s.StartDecisionTick() + s.startEffortBackfill(runner) } return err } +// cancelEffortBackfill cancels a running effort backfill. A run cancelled +// before its last chunk writes no marker, so the next run redoes it; one +// cancelled after it may still mark, which is true, since every candidate was +// enqueued. +func (s *Server) cancelEffortBackfill() { + s.decisionTick.mu.Lock() + defer s.decisionTick.mu.Unlock() + if s.decisionTick.backfillCancel != nil { + s.decisionTick.backfillCancel() + s.decisionTick.backfillCancel = nil + } +} + +// startEffortBackfill runs the one-shot effort backfill (TASK-3921) off the +// caller's path. It sits here, not at boot, so it covers a provider the admin +// enables LATER as well as one configured at startup; the marker makes every +// call after the first complete one a cheap read. A new start cancels the run +// before it (reconfigure builds a new runner every time), and a failed run, +// or one cancelled before its last chunk, writes no marker and is retried by +// the next reconfigure or restart. It makes no provider call; it only owes jobs, which the tick +// answers under the debounce. The stopped check and bg.Add share one hold of +// decisionTick.mu, so Stop() either precedes both (nothing starts) or follows +// both (bg.Wait drains it and stopDecisionTick cancels it). +func (s *Server) startEffortBackfill(runner *decision.Runner) { + s.decisionTick.mu.Lock() + if s.decisionTick.stopped || runner == nil { + s.decisionTick.mu.Unlock() + return + } + if s.decisionTick.backfillCancel != nil { + s.decisionTick.backfillCancel() + } + ctx, cancel := context.WithCancel(context.Background()) + s.decisionTick.backfillCancel = cancel + s.decisionTick.backfillGen++ + gen := s.decisionTick.backfillGen + s.bg.Add(1) + s.decisionTick.mu.Unlock() + go func() { + defer s.bg.Done() + defer func() { + cancel() + s.decisionTick.mu.Lock() + if s.decisionTick.backfillGen == gen { + s.decisionTick.backfillCancel = nil + } + s.decisionTick.mu.Unlock() + }() + defer s.recoverSweeper("effort-backfill") + if _, err := runner.BackfillEffort(ctx); err != nil && ctx.Err() == nil { + slog.Error("effort backfill failed; it will be retried on the next restart or settings change", "error", err) + } + }() +} + // resolveDecisionConfigLocked returns the resolved config, the origin of each // field, and any error reading the stored secret. A stored key that cannot be // read — it does not decrypt, or the row is not ciphertext at all — is treated diff --git a/internal/server/decision_settings_test.go b/internal/server/decision_settings_test.go index 1cb82e806..e2825567e 100644 --- a/internal/server/decision_settings_test.go +++ b/internal/server/decision_settings_test.go @@ -574,3 +574,53 @@ func (p *countingProvider) Ask(_ context.Context, _ any, qs map[string]decision. } return out, decision.Usage{}, nil } + +// TASK-3921, from the consuming side (CONVE-19): the effort backfill must run +// when an admin enables the provider AFTER boot, not only at boot. Boot with no +// provider, seed an item the attention set has answered for needs_human_decision +// but never for effort, enable a provider through the admin endpoint, and see +// the job owed and the marker set. A boot-only hook never runs here, so this +// fails against one. +func TestConfigureDecisions_BackfillsOnLaterEnable(t *testing.T) { + clearDecisionEnv(t) + srv, token := decisionSettingsServer(t) + if srv.decisionRunner() != nil { + t.Fatal("precondition: a runner is attached at boot") + } + ws, coll := decisionSettingsWorkspace(t, srv) + item, err := srv.store.CreateItem(ws.ID, coll.ID, models.ItemCreate{Title: "Pick a vendor", Fields: `{"status":"open"}`}) + if err != nil { + t.Fatal(err) + } + if err := srv.store.InsertItemDecisions(ws.ID, []models.ItemDecision{{ + ItemID: item.ID, QuestionSet: decision.AttentionSetName, QuestionKey: decision.AttentionNeedsHuman, Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.9}`), Provider: "fake", Model: "m", + StateHash: "h", QuestionHash: "q", ItemSeq: 1, + }}); err != nil { + t.Fatal(err) + } + if j, _ := srv.store.GetDecisionJob(item.ID, decision.AttentionSetName); j != nil { + t.Fatal("precondition: a job is owed with decisions off") + } + if _, code, body := putDecisionSettings(t, srv, token, map[string]interface{}{"enabled": true, "api_key": decisionTestKey}); code != http.StatusOK { + t.Fatalf("enable: %d %s", code, body) + } + deadline := time.Now().Add(10 * time.Second) + for { + j, err := srv.store.GetDecisionJob(item.ID, decision.AttentionSetName) + if err != nil { + t.Fatal(err) + } + marker, err := srv.store.GetPlatformSetting(decision.EffortBackfillSetting) + if err != nil { + t.Fatal(err) + } + if j != nil && marker != "" { + return + } + if time.Now().After(deadline) { + t.Fatalf("after enabling: job owed = %v, marker = %q; the backfill did not run on a later enable", j != nil, marker) + } + time.Sleep(20 * time.Millisecond) + } +} diff --git a/internal/server/decision_tick.go b/internal/server/decision_tick.go index 9a3eee221..c1fafbbc2 100644 --- a/internal/server/decision_tick.go +++ b/internal/server/decision_tick.go @@ -78,6 +78,13 @@ type decisionTickConfig struct { // (codex round 1). A cancelled job's claim is released without counting // an attempt, so the new runner picks it up. passCancel context.CancelFunc + + // backfillCancel cancels the running effort backfill (TASK-3921), if any; + // backfillGen names the current run so a cancelled run exiting late cannot + // clear the state of the one that replaced it. Both live under mu, as the + // stopped check and the bg.Add of the run do. + backfillCancel context.CancelFunc + backfillGen uint64 } // SetDecisionRunner attaches the decision runner and wires its registry into @@ -179,6 +186,10 @@ func (s *Server) stopDecisionTick() { s.decisionTick.mu.Lock() defer s.decisionTick.mu.Unlock() s.decisionTick.stopped = true + if s.decisionTick.backfillCancel != nil { + s.decisionTick.backfillCancel() + s.decisionTick.backfillCancel = nil + } if !s.decisionTick.running { return } diff --git a/internal/server/effort_backfill_lifecycle_test.go b/internal/server/effort_backfill_lifecycle_test.go new file mode 100644 index 000000000..ebe84e34d --- /dev/null +++ b/internal/server/effort_backfill_lifecycle_test.go @@ -0,0 +1,218 @@ +package server + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/decision" + "github.com/PerpetualSoftware/pad/internal/models" +) + +// backfillRig is a server with two backfill candidates, decisions off, and a +// hook that parks the FIRST run at chunk 0 (chunk size 1) until it is +// cancelled or released. Later runs pass straight through. +type backfillRig struct { + srv *Server + items []*models.Item + entered chan struct{} // first run reached its first chunk + cancelled chan struct{} // first run saw its context cancelled + release chan struct{} + runs atomic.Int32 + hookCalls atomic.Int32 + once sync.Once +} + +func newBackfillRig(t *testing.T) *backfillRig { + t.Helper() + clearDecisionEnv(t) + srv, _ := decisionSettingsServer(t) + ws, coll := decisionSettingsWorkspace(t, srv) + g := &backfillRig{srv: srv, entered: make(chan struct{}), cancelled: make(chan struct{}), release: make(chan struct{})} + for _, title := range []string{"One", "Two"} { + it, err := srv.store.CreateItem(ws.ID, coll.ID, models.ItemCreate{Title: title, Fields: `{"status":"open"}`}) + if err != nil { + t.Fatal(err) + } + if err := srv.store.InsertItemDecisions(ws.ID, []models.ItemDecision{{ + ItemID: it.ID, QuestionSet: decision.AttentionSetName, QuestionKey: decision.AttentionNeedsHuman, Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.9}`), Provider: "fake", Model: "m", + StateHash: "h", QuestionHash: "q", ItemSeq: 1, + }}); err != nil { + t.Fatal(err) + } + g.items = append(g.items, it) + } + srv.decisionSettings.backfillChunk = 1 + srv.decisionSettings.backfillHook = func(ctx context.Context, chunk int) error { + g.hookCalls.Add(1) + if chunk != 0 || g.runs.Add(1) != 1 { + return nil + } + close(g.entered) + select { + case <-ctx.Done(): + close(g.cancelled) + return ctx.Err() + case <-g.release: + return nil + } + } + t.Cleanup(func() { g.once.Do(func() { close(g.release) }) }) + t.Setenv("PAD_DECISION_PROVIDER", decision.ProviderTypesafe) + t.Setenv("PAD_TYPESAFE_API_KEY", "env-key") + return g +} + +func (g *backfillRig) start(t *testing.T) { + t.Helper() + if err := g.srv.ConfigureDecisions(); err != nil { + t.Fatal(err) + } + select { + case <-g.entered: + case <-time.After(10 * time.Second): + t.Fatal("the backfill never reached its first chunk") + } +} + +func (g *backfillRig) waitCancelled(t *testing.T, what string) { + t.Helper() + select { + case <-g.cancelled: + case <-time.After(5 * time.Second): + t.Fatalf("%s did not cancel the running backfill", what) + } +} + +func (g *backfillRig) drain(t *testing.T) { + t.Helper() + g.once.Do(func() { close(g.release) }) + g.srv.stopDecisionTick() // the tick loop is on bg too + done := make(chan struct{}) + go func() { g.srv.bg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("a background goroutine did not exit") + } +} + +func (g *backfillRig) owedJobs(t *testing.T) int { + t.Helper() + n := 0 + for _, it := range g.items { + j, err := g.srv.store.GetDecisionJob(it.ID, decision.AttentionSetName) + if err != nil { + t.Fatal(err) + } + if j != nil { + n++ + } + } + return n +} + +func (g *backfillRig) marker(t *testing.T) string { + t.Helper() + m, err := g.srv.store.GetPlatformSetting(decision.EffortBackfillSetting) + if err != nil { + t.Fatal(err) + } + return m +} + +// F1: Stop() while a backfill is parked mid-run cancels it, the goroutine +// exits (bg drains), and the marker is never written. +func TestEffortBackfill_StopCancelsARunningBackfill(t *testing.T) { + g := newBackfillRig(t) + g.start(t) + g.srv.stopDecisionTick() + g.waitCancelled(t, "stopDecisionTick") + g.drain(t) + if m := g.marker(t); m != "" { + t.Errorf("marker written by a cancelled backfill: %q", m) + } + if n := g.owedJobs(t); n != 0 { + t.Errorf("%d jobs owed by a backfill cancelled before its first chunk", n) + } +} + +// F1: after Stop() a start does nothing: no run, no state left behind. +func TestEffortBackfill_NoStartAfterStop(t *testing.T) { + g := newBackfillRig(t) + g.srv.stopDecisionTick() + if err := g.srv.ConfigureDecisions(); err != nil { + t.Fatal(err) + } + r := g.srv.decisionRunner() + if r == nil { + t.Fatal("precondition: no runner, so a start was never attempted") + } + g.srv.startEffortBackfill(r) + g.drain(t) + g.srv.decisionTick.mu.Lock() + gen, c := g.srv.decisionTick.backfillGen, g.srv.decisionTick.backfillCancel + g.srv.decisionTick.mu.Unlock() + if gen != 0 || c != nil || g.hookCalls.Load() != 0 { + t.Errorf("a backfill started after Stop: gen=%d cancel=%v hook calls=%d", gen, c != nil, g.hookCalls.Load()) + } + if m := g.marker(t); m != "" || g.owedJobs(t) != 0 { + t.Errorf("a backfill ran after Stop: marker %q jobs %d", m, g.owedJobs(t)) + } +} + +// F2: disabling decisions while a backfill is parked before its first chunk +// cancels it: no job is owed (rows counted, not just the marker) and no +// marker is written. A chunk already inside OweDecisionJobs when the disable +// lands may still commit (lead ruling bdce2116, docs only); this test does +// not cover that window. +func TestEffortBackfill_DisableCancelsARunningBackfill(t *testing.T) { + g := newBackfillRig(t) + g.start(t) + t.Setenv("PAD_DECISION_PROVIDER", decision.ProviderNone) + if err := g.srv.ConfigureDecisions(); err != nil { + t.Fatal(err) + } + if g.srv.decisionRunner() != nil { + t.Fatal("precondition: decisions are still on") + } + g.waitCancelled(t, "disabling") + g.drain(t) + if n := g.owedJobs(t); n != 0 { + t.Errorf("%d decision_jobs rows owed after a disable, with no provider", n) + } + if m := g.marker(t); m != "" { + t.Errorf("marker written after a disable: %q", m) + } +} + +// F2: replacing the runner mid-backfill cancels the old run, and the new run +// completes: both items owed, marker set, the old run having written nothing. +func TestEffortBackfill_ReplaceCancelsTheOldRunAndTheNewOneCompletes(t *testing.T) { + g := newBackfillRig(t) + g.start(t) + if err := g.srv.ConfigureDecisions(); err != nil { // new runner, same provider + t.Fatal(err) + } + g.waitCancelled(t, "replacing the runner") + deadline := time.Now().Add(10 * time.Second) + for g.marker(t) == "" { + if time.Now().After(deadline) { + t.Fatal("the replacement backfill never set the marker") + } + time.Sleep(20 * time.Millisecond) + } + g.drain(t) + if n := g.owedJobs(t); n != 2 { + t.Errorf("owed jobs = %d, want both items", n) + } + g.srv.decisionTick.mu.Lock() + c := g.srv.decisionTick.backfillCancel + g.srv.decisionTick.mu.Unlock() + if c != nil { + t.Error("the finished run left its cancel registered") + } +} diff --git a/internal/server/handlers_items.go b/internal/server/handlers_items.go index c7dda33bc..e4c16034d 100644 --- a/internal/server/handlers_items.go +++ b/internal/server/handlers_items.go @@ -1304,7 +1304,7 @@ func (s *Server) handleGetItem(w http.ResponseWriter, r *http.Request) { // failure here is logged and the member omitted rather than failing the // item read: decisions are advisory, and an item that cannot be fetched // because a secondary read failed is a worse answer than one without them. - if decisions, derr := s.decisionRunner().Decisions(item.ID); derr != nil { + if decisions, derr := s.decisionRunner().ItemReadDecisions(item.ID); derr != nil { slog.Warn("item read: decisions unavailable", "item_id", item.ID, "error", derr) } else if len(decisions) > 0 { item.Decisions = decisions diff --git a/internal/server/item_show_staged_decisions_test.go b/internal/server/item_show_staged_decisions_test.go new file mode 100644 index 000000000..9da4fdafd --- /dev/null +++ b/internal/server/item_show_staged_decisions_test.go @@ -0,0 +1,67 @@ +package server + +import ( + "strings" + "testing" + + "github.com/PerpetualSoftware/pad/internal/decision" + "github.com/PerpetualSoftware/pad/internal/models" +) + +// TASK-3921 F3, through the router (CONVE-19): item show carries +// needs_human_decision and none of the staged keys, while the decisions +// endpoint returns all four. +func TestItemShow_OmitsStagedDecisionKeys(t *testing.T) { + t.Parallel() + srv := testServer(t) + reg, err := decision.ProductionRegistry() + if err != nil { + t.Fatal(err) + } + srv.SetDecisionRunner(decision.NewRunner(srv.store, &stubProvider{}, reg)) + + slug := createWSWithCollections(t, srv) + item := createItem(t, srv, slug, "tasks", map[string]interface{}{"title": "Staged", "fields": `{"status":"open"}`}) + ws, err := srv.store.GetWorkspaceBySlug(slug) + if err != nil || ws == nil { + t.Fatalf("workspace: %v", err) + } + keys := []string{decision.AttentionNeedsHuman, decision.AttentionEffort, "effort_medium", decision.AttentionHighStakes} + var rows []models.ItemDecision + for _, k := range keys { + rows = append(rows, models.ItemDecision{ + ItemID: item.ID, QuestionSet: decision.AttentionSetName, QuestionKey: k, Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.9}`), Provider: "fake", Model: "m", + StateHash: "h", QuestionHash: "q-" + k, ItemSeq: 1, + }) + } + if err := srv.store.InsertItemDecisions(ws.ID, rows); err != nil { + t.Fatal(err) + } + + show := itemJSONKeys(t, srv, slug, item.Slug) + raw, ok := show["decisions"] + if !ok { + t.Fatal("item show carries no decisions member") + } + got := string(raw) + if !strings.Contains(got, decision.AttentionNeedsHuman) { + t.Errorf("item show lost needs_human_decision: %s", got) + } + for _, k := range keys[1:] { + if strings.Contains(got, `"`+k+`"`) { + t.Errorf("item show carries staged key %s: %s", k, got) + } + } + + all := getDecisions(t, srv, slug, item.Slug) + seen := map[string]bool{} + for _, d := range all.Decisions { + seen[d.QuestionKey] = true + } + for _, k := range keys { + if !seen[k] { + t.Errorf("the decisions endpoint is missing %s (it is RAW by design)", k) + } + } +} diff --git a/internal/store/decisions.go b/internal/store/decisions.go index f5f05d9e3..0cedfde0a 100644 --- a/internal/store/decisions.go +++ b/internal/store/decisions.go @@ -172,29 +172,113 @@ func (s *Store) enqueueDecisionJobsTx(tx *sql.Tx, workspaceID, itemID, collectio now := time.Now() ts := decisionTime(now) for _, set := range sets { - notBefore := "" - if window := decisionWindowFor(debounce, set); window > 0 { - var err error - if notBefore, err = decisionNotBefore(tx.QueryRow(s.q(latestDecisionEvaluationSQL), itemID, set), window, now); err != nil { - return fmt.Errorf("enqueue decision job %s/%s: %w", itemID, set, err) - } - } - if _, err := tx.Exec(s.q(` - INSERT INTO decision_jobs (item_id, workspace_id, question_set, generation, enqueued_at, attempts, not_before) - VALUES (?, ?, ?, 1, ?, 0, ?) - ON CONFLICT (item_id, question_set) DO UPDATE SET - generation = decision_jobs.generation + 1, - attempts = 0, - last_error = NULL, - not_before = COALESCE(decision_jobs.not_before, excluded.not_before)`), - itemID, workspaceID, set, ts, nullableText(notBefore), - ); err != nil { - return fmt.Errorf("enqueue decision job %s/%s: %w", itemID, set, err) + if err := s.owedDecisionJobTx(tx, workspaceID, itemID, set, debounce, now, ts); err != nil { + return err } } return nil } +// owedDecisionJobTx is the per-(item, set) body of enqueueDecisionJobsTx, +// shared with OweDecisionJobs so the backfill and the write doors cannot +// drift: same upsert, same coalescing, same debounce. +func (s *Store) owedDecisionJobTx(tx *sql.Tx, workspaceID, itemID, set string, debounce DecisionDebounce, now time.Time, ts string) error { + notBefore := "" + if window := decisionWindowFor(debounce, set); window > 0 { + var err error + if notBefore, err = decisionNotBefore(tx.QueryRow(s.q(latestDecisionEvaluationSQL), itemID, set), window, now); err != nil { + return fmt.Errorf("enqueue decision job %s/%s: %w", itemID, set, err) + } + } + if _, err := tx.Exec(s.q(` + INSERT INTO decision_jobs (item_id, workspace_id, question_set, generation, enqueued_at, attempts, not_before) + VALUES (?, ?, ?, 1, ?, 0, ?) + ON CONFLICT (item_id, question_set) DO UPDATE SET + generation = decision_jobs.generation + 1, + attempts = 0, + last_error = NULL, + not_before = COALESCE(decision_jobs.not_before, excluded.not_before)`), + itemID, workspaceID, set, ts, nullableText(notBefore), + ); err != nil { + return fmt.Errorf("enqueue decision job %s/%s: %w", itemID, set, err) + } + return nil +} + +// oweDecisionChunk bounds the items one OweDecisionJobs transaction covers. +const oweDecisionChunk = 500 + +// OweDecisionJobs makes the one question set owed for each item, in chunked +// transactions, under the installed debounce (an item evaluated inside the +// window gets a future not_before, exactly as a write would give it). It does +// not consult the set resolver or any eligibility: the caller has decided. +// Re-owing an item that already has a job bumps its generation and keeps one +// row. Used by the one-shot effort backfill (TASK-3921). +func (s *Store) OweDecisionJobs(workspaceID, set string, itemIDs []string) error { + debounce := s.decisionDebounceFunc() + for start := 0; start < len(itemIDs); start += oweDecisionChunk { + end := start + oweDecisionChunk + if end > len(itemIDs) { + end = len(itemIDs) + } + if err := s.oweDecisionChunk(workspaceID, set, itemIDs[start:end], debounce); err != nil { + return err + } + } + return nil +} + +func (s *Store) oweDecisionChunk(workspaceID, set string, itemIDs []string, debounce DecisionDebounce) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck // no-op after commit + now := time.Now() + ts := decisionTime(now) + for _, id := range itemIDs { + if err := s.owedDecisionJobTx(tx, workspaceID, id, set, debounce, now, ts); err != nil { + return err + } + } + return tx.Commit() +} + +// DecisionBackfillCandidate is one live item the backfill may owe a job for. +type DecisionBackfillCandidate struct { + ItemID string + WorkspaceID string +} + +// ItemsMissingDecisionKey lists the live items that have a stored answer for +// withKey in the set (in any state or hash) and NONE for missingKey: the items +// the set has been asked about before the question missingKey existed. The +// caller filters by the set's current eligibility. Ordered for stable chunks. +func (s *Store) ItemsMissingDecisionKey(set, withKey, missingKey string) ([]DecisionBackfillCandidate, error) { + rows, err := s.db.Query(s.q(` + SELECT i.workspace_id, i.id + FROM items i + WHERE i.deleted_at IS NULL + AND EXISTS (SELECT 1 FROM item_decisions d + WHERE d.item_id = i.id AND d.question_set = ? AND d.question_key = ?) + AND NOT EXISTS (SELECT 1 FROM item_decisions e + WHERE e.item_id = i.id AND e.question_set = ? AND e.question_key = ?) + ORDER BY i.workspace_id, i.id`), set, withKey, set, missingKey) + if err != nil { + return nil, fmt.Errorf("list decision backfill candidates: %w", err) + } + defer rows.Close() + var out []DecisionBackfillCandidate + for rows.Next() { + var c DecisionBackfillCandidate + if err := rows.Scan(&c.WorkspaceID, &c.ItemID); err != nil { + return nil, err + } + out = append(out, c) + } + return out, rows.Err() +} + // enqueueDecisionJobsForItemTx is the comment door's form: a comment knows its // item but not the item's collection. func (s *Store) enqueueDecisionJobsForItemTx(tx *sql.Tx, itemID string) error { @@ -293,6 +377,41 @@ func (s *Store) ClaimDecisionJobs(runner string, limit int, lease time.Duration) return claimed, nil } +// CountClaimableDecisionJobs counts the set's jobs for itemIDs that +// ClaimDecisionJobs could take RIGHT NOW at `now`: the very decisionJobClaimable +// predicate, so the two cannot drift. A job that is claimed (lease live), held +// by the re-ask debounce (not_before ahead), held by a soft-deleted workspace, +// answered or deleted is not counted. The effort backfill uses it to tell +// whether its last chunk has drained (TASK-3921). +func (s *Store) CountClaimableDecisionJobs(set string, itemIDs []string, now time.Time) (int, error) { + nowS := decisionTime(now) + total := 0 + for start := 0; start < len(itemIDs); start += oweDecisionChunk { + end := start + oweDecisionChunk + if end > len(itemIDs) { + end = len(itemIDs) + } + part := itemIDs[start:end] + args := make([]any, 0, len(part)+3) + args = append(args, set) + marks := make([]string, len(part)) + for i, id := range part { + marks[i] = "?" + args = append(args, id) + } + args = append(args, nowS, nowS) + var n int + if err := s.db.QueryRow(s.q(` + SELECT COUNT(*) FROM decision_jobs + WHERE question_set = ? AND item_id IN (`+strings.Join(marks, ",")+`) + AND `+decisionJobClaimable), args...).Scan(&n); err != nil { + return 0, fmt.Errorf("count claimable decision jobs: %w", err) + } + total += n + } + return total, nil +} + // CompleteDecisionJob retires a claimed job. // // The job is deleted only if its generation is still the one claimed. If a diff --git a/internal/store/decisions_owe_test.go b/internal/store/decisions_owe_test.go new file mode 100644 index 000000000..2cbe57eb8 --- /dev/null +++ b/internal/store/decisions_owe_test.go @@ -0,0 +1,144 @@ +package store + +import ( + "github.com/PerpetualSoftware/pad/internal/models" + "testing" + "time" +) + +// TASK-3921: OweDecisionJobs and the backfill candidate query. Both backends +// via testStore. + +func TestOweDecisionJobs_DebounceAndCoalescing(t *testing.T) { + s, ws, col := debounceFixture(t) + // Install a resolver that owes nothing, so the creations below leave no jobs. + s.SetDecisionSetResolver(func(string) []string { return nil }) + recent := createTestItem(t, s, ws.ID, col.ID, "Recent", "") + old := createTestItem(t, s, ws.ID, col.ID, "Old", "") + fresh := createTestItem(t, s, ws.ID, col.ID, "Fresh", "") + evaluated(t, s, ws, recent.ID, "triage", time.Now().Add(-10*time.Minute)) + evaluated(t, s, ws, old.ID, "triage", time.Now().Add(-2*time.Hour)) + if err := s.OweDecisionJobs(ws.ID, "triage", []string{recent.ID, old.ID, fresh.ID}); err != nil { + t.Fatal(err) + } + if j := mustJob(t, s, recent.ID, "triage"); j.NotBefore == "" { + t.Error("an item evaluated 10m ago was owed with no not_before") + } + if j := mustJob(t, s, old.ID, "triage"); j.NotBefore != "" { + t.Errorf("an item evaluated 2h ago got not_before %q", j.NotBefore) + } + if j := mustJob(t, s, fresh.ID, "triage"); j.NotBefore != "" { + t.Errorf("a never-evaluated item got not_before %q", j.NotBefore) + } + // Owing again coalesces into the same row. + if err := s.OweDecisionJobs(ws.ID, "triage", []string{recent.ID}); err != nil { + t.Fatal(err) + } + if n := countDecisionJobs(t, s); n != 3 { + t.Errorf("%d job rows after a repeat; want 3", n) + } + if j := mustJob(t, s, recent.ID, "triage"); j.Generation != 2 { + t.Errorf("generation = %d; want 2", j.Generation) + } +} + +func TestItemsMissingDecisionKey(t *testing.T) { + s, ws, col := decisionFixture(t) + ask := func(itemID string, keys ...string) { + for _, k := range keys { + if err := s.InsertItemDecisions(ws.ID, []models.ItemDecision{{ + ItemID: itemID, QuestionSet: "attention", QuestionKey: k, Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.5}`), Provider: "fake", Model: "m", + StateHash: "h", QuestionHash: "q", ItemSeq: 1, EvaluatedAt: decisionTime(time.Now()), + }}); err != nil { + t.Fatal(err) + } + } + } + onlyNeeds := createTestItem(t, s, ws.ID, col.ID, "A", "") + both := createTestItem(t, s, ws.ID, col.ID, "B", "") + none := createTestItem(t, s, ws.ID, col.ID, "C", "") + gone := createTestItem(t, s, ws.ID, col.ID, "D", "") + otherSet := createTestItem(t, s, ws.ID, col.ID, "E", "") + ask(onlyNeeds.ID, "needs_human_decision") + ask(both.ID, "needs_human_decision", "effort") + ask(gone.ID, "needs_human_decision") + if err := s.InsertItemDecisions(ws.ID, []models.ItemDecision{{ + ItemID: otherSet.ID, QuestionSet: "conventions", QuestionKey: "needs_human_decision", Kind: "noul", + Answer: []byte(`{"type":"noul","noul":0.5}`), Provider: "fake", Model: "m", + StateHash: "h", QuestionHash: "q", ItemSeq: 1, EvaluatedAt: decisionTime(time.Now()), + }}); err != nil { + t.Fatal(err) + } + if err := s.DeleteItem(gone.ID); err != nil { + t.Fatal(err) + } + _ = none + got, err := s.ItemsMissingDecisionKey("attention", "needs_human_decision", "effort") + if err != nil { + t.Fatal(err) + } + if len(got) != 1 || got[0].ItemID != onlyNeeds.ID || got[0].WorkspaceID != ws.ID { + t.Fatalf("candidates = %+v; want only item A", got) + } +} + +// CountClaimableDecisionJobs counts what ClaimDecisionJobs could take now: not +// a claimed job, not a debounced one, not a retired one, nor another item's. +func TestCountClaimableDecisionJobs(t *testing.T) { + s, ws, col := debounceFixture(t) + s.SetDecisionSetResolver(func(string) []string { return nil }) + recent := createTestItem(t, s, ws.ID, col.ID, "Recent", "") + a := createTestItem(t, s, ws.ID, col.ID, "A", "") + b := createTestItem(t, s, ws.ID, col.ID, "B", "") + outside := createTestItem(t, s, ws.ID, col.ID, "Outside", "") + evaluated(t, s, ws, recent.ID, "triage", time.Now().Add(-10*time.Minute)) + if err := s.OweDecisionJobs(ws.ID, "triage", []string{recent.ID, a.ID, b.ID, outside.ID}); err != nil { + t.Fatal(err) + } + ids := []string{recent.ID, a.ID, b.ID} + count := func(set string, at time.Time) int { + t.Helper() + n, err := s.CountClaimableDecisionJobs(set, ids, at) + if err != nil { + t.Fatal(err) + } + return n + } + now := time.Now() + // recent is debounced; outside is not asked about; the other set has none. + if n := count("triage", now); n != 2 { + t.Fatalf("claimable = %d; want 2 (a, b; recent is debounced)", n) + } + if n := count("other", now); n != 0 { + t.Fatalf("another set counted %d", n) + } + if n, err := s.CountClaimableDecisionJobs("triage", nil, now); err != nil || n != 0 { + t.Fatalf("no ids = %d, %v", n, err) + } + // Claim everything claimable (a, b, outside): none is claimable now. + jobs, err := s.ClaimDecisionJobs("r", 10, time.Hour) + if err != nil || len(jobs) != 3 { + t.Fatalf("claim: %v %v", jobs, err) + } + if n := count("triage", now); n != 0 { + t.Fatalf("after claiming claimable = %d; want 0 (claimed and debounced)", n) + } + // A lease that has run out makes the claimed jobs claimable again (a, b). + if n := count("triage", now.Add(2*time.Hour)); n != 3 { + t.Fatalf("expired leases + passed debounce claimable = %d; want 3", n) + } + // Retire them: none claimable now. + for _, j := range jobs { + if err := s.CompleteDecisionJob(j); err != nil { + t.Fatal(err) + } + } + if n := count("triage", now); n != 0 { + t.Fatalf("after completing claimable = %d; want 0", n) + } + // Two hours on, only recent (debounce passed) remains. + if n := count("triage", now.Add(2*time.Hour)); n != 1 { + t.Fatalf("two hours on claimable = %d; want 1", n) + } +} diff --git a/scripts/decision-eval/attention/RESULTS.md b/scripts/decision-eval/attention/RESULTS.md index 8b6c84c46..a57272bd0 100644 --- a/scripts/decision-eval/attention/RESULTS.md +++ b/scripts/decision-eval/attention/RESULTS.md @@ -68,3 +68,44 @@ results-decision.json) and `items/.json` + `items/.comments.json` from `pad item show|comments --format json`. `-dry` builds every state without calling the provider, and must reproduce the day-73 AUC of 0.939 from the recorded probabilities. + +## TASK-3957 and TASK-3921 (2026-10-19) + +The same 140 items and labels, with the dump rebuilt from the day-73 +results-decision.json and the live items. `-dry` reproduces 0.939, and every +run had 0 errors. r1 excludes the same 2 positives as above (closed with no +logged terminal transition). Model `jev-1.13.0`. Only needs_human_decision is +scored. Each run prints the question keys it sends (`questions per call=`). + +- TASK-3957: the ONE-question call after TASK-3915 (`blocked` and + `waiting_on_external` retired), run at e432d0f5b. +- TASK-3921: the FOUR-question call, adding the staged effort, effort_medium + and high_stakes. +- `titlebody` is unchanged: it sends the day-73 pair (needs_human_decision + + `blocked`) and remains the model-drift control. +- `titlebody-live` (new) sends the live set on the same {title, body} state, so + it measures the call shape without the production state. The TASK-3921 gate + is needs_human_decision within .02 of 0.939 there. + +| mode | questions per call | AUC | P / R at 0.5 | P / R at 0.7 | +|---|---|---|---|---| +| titlebody (drift control) | 2 (day-73 pair) | 0.934 | 0.927 / 0.729 | 0.919 / 0.486 | +| production, TASK-3957 | 1 | 0.871 | 0.917 / 0.314 | 0.818 / 0.129 | +| r1, TASK-3957 | 1 | 0.853 | 0.794 / 0.735 | 0.791 / 0.500 | +| titlebody-live, TASK-3921 (gate) | 4 | 0.936 | 0.944 / 0.729 | 0.944 / 0.486 | +| production, TASK-3921 | 4 | 0.865 | 0.923 / 0.343 | 0.800 / 0.114 | +| r1, TASK-3921 | 4 | 0.856 | 0.810 / 0.750 | 0.775 / 0.456 | + +Reading: + +- No model drift: the control gives 0.934, against 0.935 on 09-22. +- The gate passes. titlebody-live gives 0.936, 0.003 below 0.939, and is not + measurably different from the two-question control. +- Adding the three staged questions does not move needs_human_decision on the + production or replayed state: production went 0.871 to 0.865 and r1 went + 0.853 to 0.856. +- r1 is 0.03 below its 09-22 figure (0.885, three questions) under both the one- + and four-question calls, so it predates this change. These runs cannot say + whether the TASK-3915 call shape or another runner change since 09-22 + opened it. Production compares only within this table, because its + states are live and have moved since 09-22. diff --git a/scripts/decision-eval/attention/main.go b/scripts/decision-eval/attention/main.go index 97557c20b..3e69f9039 100644 --- a/scripts/decision-eval/attention/main.go +++ b/scripts/decision-eval/attention/main.go @@ -16,6 +16,10 @@ // -mode titlebody state = {"title","body"[:5000 runes]} with the same two // eval questions — the day-73 shape, re-run today, as the // control that separates model drift from state shape. +// -mode titlebody-live the same {"title","body"} state with the LIVE +// attention set (four questions since TASK-3921) in one +// call: what the call shape does to needs_human_decision on +// the day-73 state, with titlebody as its drift control. // // The key is read from PAD_TYPESAFE_API_KEY or TYPESAFE_API_KEY and is never // printed or written. @@ -70,7 +74,7 @@ type Row struct { func main() { dump := flag.String("dump", "", "directory holding population.json and items/") - mode := flag.String("mode", "production", "production | titlebody") + mode := flag.String("mode", "production", "production | titlebody | titlebody-live") out := flag.String("out", "", "run JSON path") dry := flag.Bool("dry", false, "build every state and report its shape; no provider call") flag.Parse() @@ -78,6 +82,24 @@ func main() { fmt.Fprintln(os.Stderr, "need -dump and -out") os.Exit(2) } + all := decision.AttentionSet().Questions + questions := all + if *mode == "titlebody" { + questions = map[string]decision.Question{ + decision.AttentionNeedsHuman: all[decision.AttentionNeedsHuman], + day73Blocked: day73BlockedQuestion, + } + } else if *mode != "titlebody-live" && *mode != "production" && *mode != "nostatus" && *mode != "nostatus-notrail" && *mode != "r0" && *mode != "r1" { + fmt.Fprintln(os.Stderr, "unknown -mode", *mode) + os.Exit(2) + } + qkeys := make([]string, 0, len(questions)) + for k := range questions { + qkeys = append(qkeys, k) + } + sort.Strings(qkeys) + fmt.Printf("mode=%s questions per call=%d: %s\n", *mode, len(qkeys), strings.Join(qkeys, ", ")) + if *dry { if *mode == "r0" || *mode == "r1" { replayDry(*dump, *mode) @@ -98,18 +120,6 @@ func main() { var pop []member mustRead(filepath.Join(*dump, "population.json"), &pop) - all := decision.AttentionSet().Questions - questions := all - if *mode == "titlebody" { - questions = map[string]decision.Question{ - decision.AttentionNeedsHuman: all[decision.AttentionNeedsHuman], - day73Blocked: day73BlockedQuestion, - } - } else if *mode != "production" && *mode != "nostatus" && *mode != "nostatus-notrail" && *mode != "r0" && *mode != "r1" { - fmt.Fprintln(os.Stderr, "unknown -mode", *mode) - os.Exit(2) - } - rows := make([]Row, len(pop)) var wg sync.WaitGroup sem := make(chan struct{}, 6) @@ -198,7 +208,7 @@ func evalOne(p decision.Provider, dump, mode string, m member, qs map[string]dec } state, _ = json.Marshal(is) row.Truncated = st.Truncated - case "titlebody": + case "titlebody", "titlebody-live": body := []rune(item.Content) if len(body) > 5000 { body = body[:5000] diff --git a/web/src/lib/decisions/attentionChips.test.ts b/web/src/lib/decisions/attentionChips.test.ts index 576af8441..939b36da1 100644 --- a/web/src/lib/decisions/attentionChips.test.ts +++ b/web/src/lib/decisions/attentionChips.test.ts @@ -11,14 +11,20 @@ function row(key: string, noul: number, over: Partial = {}): ItemD } describe('attentionChips', () => { - it('labels the known key first, and flags at the dashboard threshold', () => { - const chips = attentionChips([ - row('some_future_key', 0.7), - row('needs_human_decision', 0.694), - ]); - expect(chips.map((c) => c.key)).toEqual(['needs_human_decision', 'some_future_key']); + it('labels the known key, and flags at the dashboard threshold', () => { + const chips = attentionChips([row('needs_human_decision', 0.694), row('needs_human_decision', 0.7, { id: 'b' })]); expect(chips[0]).toMatchObject({ label: 'Needs a human', percent: 69, flagged: false }); - expect(chips[1]).toMatchObject({ label: 'some_future_key', percent: 70, flagged: true }); + expect(chips[1]).toMatchObject({ label: 'Needs a human', percent: 70, flagged: true }); + }); + + it('makes no chip for a key without a label: the staged high_stakes Noul, even at .95 (TASK-3921)', () => { + const chips = attentionChips([ + row('high_stakes', 0.95), + row('some_future_key', 0.9), + row('effort', 0.9, { kind: 'score', answer: { type: 'score', score: 1 } }), + row('needs_human_decision', 0.9), + ]); + expect(chips.map((c) => c.key)).toEqual(['needs_human_decision']); }); it('shows no chip for the retired keys, whose answers the server marks not current (TASK-3915)', () => { diff --git a/web/src/lib/decisions/attentionChips.ts b/web/src/lib/decisions/attentionChips.ts index 0fd142ef5..c7d754ce4 100644 --- a/web/src/lib/decisions/attentionChips.ts +++ b/web/src/lib/decisions/attentionChips.ts @@ -8,7 +8,12 @@ import type { ItemDecision } from '$lib/types'; export const ATTENTION_SET = 'attention'; export const ATTENTION_THRESHOLD = 0.7; -// The set asks one question since TASK-3915 / TASK-3924 retired `blocked` +// Only a key with a LABEL becomes a chip (TASK-3921). The set also asks +// effort, effort_medium and high_stakes, staged before their reader +// (TASK-3959); the last is a Noul, and without this rule it would render as a +// raw-key chip. The server's brief and dashboard skip them the same way. +// +// The set asked one question after TASK-3915 / TASK-3924 retired `blocked` // (the dependency graph owns it) and `waiting_on_external`. Their stored // answers come back with current: false, so they never become chips. const LABELS: Record = { @@ -35,9 +40,11 @@ export function attentionChips(decisions: readonly ItemDecision[] | null | undef if (d.question_set !== ATTENTION_SET || d.kind !== 'noul' || !d.current) continue; const p = d.answer?.noul; if (typeof p !== 'number' || !Number.isFinite(p)) continue; + const label = LABELS[d.question_key]; + if (label === undefined) continue; chips.push({ key: d.question_key, - label: LABELS[d.question_key] ?? d.question_key, + label, percent: Math.round(p * 100), flagged: p >= ATTENTION_THRESHOLD, });