diff --git a/CAPABILITY_MAP.md b/CAPABILITY_MAP.md index 0600c5c..27cc121 100644 --- a/CAPABILITY_MAP.md +++ b/CAPABILITY_MAP.md @@ -23,7 +23,7 @@ | Recover parked or stuck Claude work | Parked notice → `Reassign to Codex` | Queued/parked items use `work:assign --feed ... --work ... --agent codex`; claimed/stuck items use claim replay by the Claude lane or `work:release --feed ... --work ... --token ...` to re-queue and rotate the token | | Schedule refresh and drain | Approve proposed cadence | `feed:heartbeat:propose`, then host `automation_update`, then `feed:heartbeat:installed` | | Verify an approved external action or cleanup | Choose the exact visible mutation, cleanup, or routine group | `action:verify` rereads the selected card-action ID, current artifact, cleanup, or batch digest immediately before mutation and records verification durably; Inbox replies additionally require `--mailbox ` to match the source message's received-at mailbox | -| Collect evidence | Render the resulting cards | `source:record-run` or `source:import-json-file` writes immutable snapshots and checkpoints; `sweep:record-batch` records the judged batch separately | +| Collect evidence | Render the resulting cards | `source:record-run` (judgments may name their `cardId`) writes immutable snapshots and, for claimed recollection work, holds the checkpoint on the run; `sweep:record-batch` records the judged batch; `sweep:status` lists what is still unpresented; `work:complete` commits the held checkpoints once every judgment is presented | | Rejudge sweep feedback | Submit dock feedback to `This sweep` | `sweep:rejudge` writes an explicit kept order and removed-card set before recollection is offered | | Render a judged item | Review its validated structured blocks, including compact two-series charts for comparative metrics | `card:upsert --feed ... --card-file ...` | | Compound learning | Answer yes when the feed thread offers its end-of-sweep learning pass; review and edit the resulting full-screen policy proposal before applying it | An idle `work:list` or `work:claim` reminds Codex to offer the pass after meaningful sweep work; `learning:request --feed ...` queues one `compound_learnings` job; Codex reviews durable evidence and uses `revision:propose --source compound`; the browser applies only the user-reviewed Markdown | diff --git a/CHANGELOG.md b/CHANGELOG.md index 09d2d65..7f964a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ a promise of ongoing maintenance. ## Unreleased +- Let sweep judgments name the card that will present them (`cardId`), so `work:complete` and the + new `tend cli sweep:status --feed ` check each judged item against its exact card and report + precisely what is still missing; judgments without a `cardId` keep the count-based check. Re-claimed + recollection work carries the same list as `operatorGuidance.pendingPresentation`, and the judge + prompt, agent contract, skill, and runbook describe the convention. Advances the CLI contract to + `0.9`. - Hold source checkpoints recorded for claimed recollection work until `work:complete` succeeds, and refuse completion until every `review` judgment has a card and every `routine_action` judgment has a card or a proposed routine action group, so a sweep diff --git a/RUNBOOK.md b/RUNBOOK.md index 81c24e2..efe1e13 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -161,6 +161,23 @@ treat broad natural-language dock input as a literal prompt edit. ## Collect +### Finish an interrupted sweep + +A sweep recorded for claimed recollection work keeps its source checkpoints on the run records until +`work:complete` succeeds, so an agent that dies after judging but before presenting loses nothing: +the next sweep starts from the previous checkpoint. To finish the interrupted one instead, re-claim +the work (the claim output carries `operatorGuidance.pendingPresentation`) or ask directly: + +```bash +tend cli sweep:status --feed +``` + +It reports `ready` plus every judgment that is not presented yet, with the run, the judgment's +position, its `cardId` when it named one, and the exact reason. Upsert the listed cards with +`sourceRunIds` including the run (or propose the routine action group), re-run `sweep:status`, then +`work:complete`. Judgments that name a `cardId` are checked against that exact card; judgments +without one are matched by count. + ### Compose meeting cards with independent readers The optional meeting-reading workflow uses ordinary feeds, source runs and cards. Follow diff --git a/docs/AGENT_CONTRACT.md b/docs/AGENT_CONTRACT.md index 79a31e2..20bea19 100644 --- a/docs/AGENT_CONTRACT.md +++ b/docs/AGENT_CONTRACT.md @@ -70,15 +70,26 @@ For `recollect_sources` work: ["INBOX"])`. Its checkpoint's `inboxEnumeration.messages` must map every authoritative `messageId` to its direct-read `threadId`; every resulting conversation must appear exactly once in `readThreadIds` or `carriedForwardThreadIds` before recording the batch. -- Record the resulting sweep with `tend cli sweep:record-batch --work `. -- Complete the work only after the source run and sweep batch are written back. -- Checkpoints recorded with `--work` are held on the run until `work:complete` succeeds, and - completion is refused until every `review` judgment in a run has its own card whose - `sourceRunIds` lists that run and that is visible for review in the current pass (or was acted - on since the run), and every `routine_action` judgment is presented by such a card or by an - item of a routine action group proposed since the sweep. If the work never completes, - the next sweep starts from the previous checkpoint and re-reads those items; a held checkpoint is - not written over one that changed after the run was recorded. +- Give every `review` or `routine_action` judgment a stable, file-safe `cardId` and reuse that + exact id in `card:upsert`. A judgment looks like + `{"decision":"review","cardId":"gmail-18c2f0a1","reason":"..."}`; `suppress` judgments need no + card. Two judgments may share one `cardId` when one card deliberately presents both. Judgments + without a `cardId` still work and are matched by count instead of by id; a card counts once per + sweep unless the judgments it presents name it. +- Record the resulting sweep with `tend cli sweep:record-batch --work `. The batch must include + every run recorded for the work (or a newer run for the same source that supersedes it). +- Upsert one card per judgment with `sourceRunIds` including its run, then run + `tend cli sweep:status --feed `; it reports `ready` and lists every judgment that is not + presented yet with the exact reason (no such card, card does not list the run, card hidden). +- Complete the work only when `sweep:status` reports `ready`. Checkpoints recorded with `--work` + are held on the run until `work:complete` succeeds, and completion is refused with the same list + otherwise. A card presents a `review` judgment only if it is individually reviewable (not part of a + routine action group) in the current pass, or was acted on since the run; `routine_action` + judgments without a card may also be covered by items of a routine action group proposed since + the sweep. If the work never completes, the next sweep + starts from the previous checkpoint and re-reads those items; a held checkpoint is not written + over one that changed after the run was recorded. Re-claiming such work returns + `operatorGuidance.pendingPresentation` with the same list. ## Core Commands @@ -103,6 +114,7 @@ Run `tend cli help` for the full command surface. Core feed-runner commands are: | Remove source | `tend cli source:remove --feed --source ` | | Record source run | `tend cli source:record-run --feed --source (--snapshots \| --snapshots-file ) (--judgments \| --judgments-file ) (--checkpoint \| --checkpoint-file ) [--context-use-file ]` | | Record sweep batch | `tend cli sweep:record-batch --feed --runs [--context ]` | +| Check what the current sweep still needs before completing | `tend cli sweep:status --feed ` | | Record sweep rejudgment | `tend cli sweep:rejudge --feed --feedback --ordered-cards --removed-cards ` | | Upsert card | `tend cli card:upsert --feed --card ` | | Run configured readers on frozen input | `tend cli readers:run --feed --run --packet-file --readers-file ` | diff --git a/docs/SKILL.md b/docs/SKILL.md index 853d12a..ed40777 100644 --- a/docs/SKILL.md +++ b/docs/SKILL.md @@ -64,8 +64,9 @@ waking this same thread and saying `go deal with the feed`. 5. Use local connectors only for the claimed item. 6. Write results back through the relevant `tend cli` command. 7. For `sweep_rejudge`, run `sweep:rejudge` against the returned `operatorGuidance.visibleCardIds` before completing the work. -8. For source recollection, record source runs and a sweep batch with the claimed `--work` id, then upsert a card for every kept judgment before completing the work. - Checkpoints recorded with `--work` only advance when `work:complete` succeeds, and completion is refused until every `review` judgment has its own card and every `routine_action` judgment has a card or a proposed routine action group. +8. For source recollection, record source runs and a sweep batch with the claimed `--work` id, then upsert a card for every `review` or `routine_action` judgment before completing the work. + Give each such judgment a stable `cardId` (for example `gmail-`) and reuse it as the card id; run `sweep:status --feed ` and complete only when it reports `ready`. + Checkpoints recorded with `--work` only advance when `work:complete` succeeds; it is refused with the list of unpresented judgments otherwise, and a re-claim returns that list as `operatorGuidance.pendingPresentation`. If context influenced collection, include a file-backed `contextUse` on the relevant source run and pin the same update id to the sweep batch. A full Gmail sweep must begin with paginated `gmail_search_email_ids(query: "", label_ids: diff --git a/server/cli/contract.ts b/server/cli/contract.ts index 173c894..30409d9 100644 --- a/server/cli/contract.ts +++ b/server/cli/contract.ts @@ -17,6 +17,7 @@ export const CLI_COMMANDS = [ "source:remove --feed --source ", "source:record-run --feed --source (--snapshots | --snapshots-file ) (--judgments | --judgments-file ) (--checkpoint | --checkpoint-file ) [--work ] [--context-use | --context-use-file ]", "sweep:record-batch --feed --runs [--work ] [--context ]", + "sweep:status --feed ", "sweep:rejudge --feed --feedback --ordered-cards --removed-cards ", "source:import-json-file --feed --source --path ", "source:import-file --feed --source --path ", diff --git a/server/cli/operator.ts b/server/cli/operator.ts index a1e2e57..5c5e685 100644 --- a/server/cli/operator.ts +++ b/server/cli/operator.ts @@ -155,6 +155,9 @@ export async function runOperatorCli(rawArgs: string[]): Promise { value("context"), ); break; + case "sweep:status": + output = await domain.sweepPresentationStatus(required("feed")); + break; case "sweep:rejudge": output = await domain.recordSweepRejudgment( required("feed"), @@ -312,11 +315,16 @@ export async function runOperatorCli(rawArgs: string[]): Promise { claimed?.kind === "execute_approved_action" ? await store.readConfig(feedId) : undefined; + const sweepPresentation = + claimed?.intent === "recollect_sources" + ? await domain.sweepPresentationStatus(feedId) + : undefined; output = formatWorkClaimOutput(feedId, work, { card, feedConfig, routineActionGroup, sweepFeedback, + sweepPresentation, }); } break; diff --git a/server/domain.ts b/server/domain.ts index 1b81f0c..8c99602 100644 --- a/server/domain.ts +++ b/server/domain.ts @@ -42,6 +42,9 @@ import type { SourceRecipe, SourceRun, SweepBatch, + SweepPresentationGap, + SweepPresentationRun, + SweepPresentationStatus, SourceRunContextUse, SweepFeedbackTrace, ThreadBinding, @@ -393,6 +396,16 @@ const CARD_BLOCK_TYPES = new Set([ const REVIEW_DECISIONS = new Set(["review", "keep"]); const ROUTINE_DECISION = "routine_action"; +// A judgment may name the card that will present it; the id must be usable as a card id so the agent +// hears about a bad id when recording the run rather than when completing the work. +function validateJudgmentCardIds(judgments: unknown[]): void { + judgments.forEach((judgment, index) => { + if (!isRecord(judgment) || judgment.cardId === undefined) return; + if (typeof judgment.cardId !== "string" || !judgment.cardId.trim()) throw new Error(`Judgment ${index + 1} cardId must be a non-empty string.`); + safeIdentifier(judgment.cardId, `Judgment ${index + 1} cardId`); + }); +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } @@ -2137,7 +2150,9 @@ export class AttentionDomain { proposedAction: input.proposedAction, items: input.items.map((item) => ({ ...item, id: item.id.trim(), title: item.title.trim(), reason: item.reason.trim() })), status: "proposed", - createdAt: existing?.createdAt ?? now, + // Every explicit upsert (only proposed, stale, or failed groups can be upserted) starts a new proposal + // cycle, so createdAt is the time of the current proposal; sweep presentation relies on that. + createdAt: now, updatedAt: now, }; for (const item of group.items) { @@ -3694,6 +3709,7 @@ export class AttentionDomain { if (!feed.sources.some((source) => source.id === sourceId)) throw new Error(`Source recipe not found: ${sourceId}`); if (triggerWorkId) await this.assertClaimedRecollectionWork(feedId, triggerWorkId); assertGmailInboxSweepEnumeration(sourceId, checkpoint); + validateJudgmentCardIds(judgments); const normalizedContextUse = contextUse ? normalizeContextUse(contextUse, await this.requireCurrentMindContext(contextUse.updateId), snapshots) : undefined; @@ -3751,6 +3767,18 @@ export class AttentionDomain { } } if (researchQuestions.size > 1) throw new Error("One sweep may originate only one On Your Mind research question."); + if (triggerWork) { + // Every run recorded for this work that still holds a checkpoint must be in the batch, or be superseded + // there by a newer run for the same source; otherwise its judged items would be neither presented nor re-read. + const included = await Promise.all(sourceRunIds.map((runId) => this.store.readRun(feedId, runId))); + const leftOut = (await this.store.listRuns(feedId)).filter((run) => + run.triggerWorkId === triggerWork.id && run.pendingCheckpoint !== undefined && !sourceRunIds.includes(run.id) + && !included.some((candidate) => candidate.sourceId === run.sourceId && (candidate.checkpointSequence ?? 0) > (run.checkpointSequence ?? 0)), + ); + if (leftOut.length) { + throw new Error(`Source ${leftOut.length === 1 ? "run" : "runs"} ${leftOut.map((run) => `${run.id} (${sourceDisplayName(run.sourceId)})`).join(", ")} recorded for this recollection ${leftOut.length === 1 ? "is" : "are"} missing from the batch. Include ${leftOut.length === 1 ? "it" : "them"}, or record a newer run for the same source and include that.`); + } + } const batchId = makeId("batch"); const supersededRoutineGroups = await this.staleProposedRoutineActionGroups(feedId, `Superseded by newer sweep batch ${batchId}.`); await this.store.writeSweepBatch({ id: batchId, feedId, sourceRunIds, ...(normalizedContextUpdateId ? { contextUpdateId: normalizedContextUpdateId } : {}), ...(triggerWorkId ? { triggerWorkId } : {}), createdAt: isoNow() }); @@ -3760,48 +3788,199 @@ export class AttentionDomain { }); } - // Recollection work completes only once every judgment that needs a card has one; that is when the - // checkpoints held on the batch's runs become the sources' live checkpoints, inside the same transaction. - // Every run is validated before anything is written so a refusal leaves both SQLite and the file mirrors untouched. - private async commitHeldSourceCheckpoints(feedId: string, batch: SweepBatch): Promise { + /** + * What the current sweep still needs before its recollection work can complete. Judgments that name a + * `cardId` are matched exactly against that card; judgments without one fall back to counting cards that + * present the run and were not claimed by an exact match. `routine_action` judgments without a card may + * also be covered by items of routine action groups proposed since the sweep. + */ + async sweepPresentationStatus(feedId: string): Promise { + return this.store.serialize(async () => { + const sweep = await this.store.readSweepState(feedId); + if (!sweep.currentBatchId) { + return { status: "idle", currentBatchId: null, workId: null, workStatus: null, ready: true, runs: [], missing: [], unbatchedRuns: [], routineGroupItems: 0, routineCoveredByGroups: 0, summary: "No sweep batch has been recorded for this feed." }; + } + return this.presentationReport(feedId, await this.store.readSweepBatch(feedId, sweep.currentBatchId)); + }); + } + + private async presentationReport(feedId: string, batch: SweepBatch): Promise { const cards = await this.store.listCards(feedId); + const cardsById = new Map(cards.map((card) => [card.id, card])); const feed = await this.store.readFeed(feedId); const workItems = await this.store.readWorkItems(feedId); const runs = await Promise.all(batch.sourceRunIds.map((runId) => this.store.readRun(feedId, runId))); - // Cards carry no judgment ids, so presentation is counted: each review judgment needs its own card that the - // user can actually see now or has already acted on since the run; routine judgments need cards or items of - // routine action groups proposed since the batch was recorded. - const presentsRun = (card: Card, run: SourceRun): boolean => { + const currentPass = feed.config.currentPass; + // A user disposition on a card (dismissal, approval, queued instruction or cleanup, reaction, block edit, + // return to review, reading preference) presents the card version that existed at the time. The feed event + // ledger is strictly ordered, so a card was acted on in its current version exactly when its latest + // disposition event comes after its latest content-write event (card:upsert); a card refreshed afterwards + // must be resurfaced. Content an agent writes while completing work the user queued or approved is the + // outcome of that reviewed item and does not count as a new write. + const CONTENT_EVENTS = new Set(["card.created", "card.updated"]); + const ACTION_EVENTS = new Set(["card.dismissed", "action.approved", "work.queued", "cleanup.queued", "voice.intent_queued", "voice.instruction_submitted", "card.reaction_recorded", "card.block_edited", "card.returned_to_review"]); + const lastWrite = new Map(); + const lastAction = new Map(); + // Routine group proposals are ordered against the batch through the same ledger: timestamps can tie within a + // millisecond, but events cannot. + const lastProposal = new Map(); + let batchRecordedAt = -1; + (await this.store.readEvents(feedId)).forEach((event, index) => { + if (event.type === "sweep.batch_recorded" && isRecord(event.detail) && event.detail.batchId === batch.id) batchRecordedAt = index; + if (event.type === "routine_action.proposed" && isRecord(event.detail) && typeof event.detail.groupId === "string") lastProposal.set(event.detail.groupId, index); + if (event.type === "reading.preference_recorded" && isRecord(event.detail) && isRecord(event.detail.beforeCardUpdatedAt)) { + for (const memberId of Object.keys(event.detail.beforeCardUpdatedAt)) lastAction.set(memberId, index); + return; + } + if (!event.cardId) return; + if (CONTENT_EVENTS.has(event.type)) lastWrite.set(event.cardId, index); + else if (ACTION_EVENTS.has(event.type)) lastAction.set(event.cardId, index); + }); + const proposedForThisSweep = (group: RoutineActionGroup): boolean => { + const proposal = lastProposal.get(group.id); + return proposal !== undefined && batchRecordedAt >= 0 ? proposal > batchRecordedAt : group.createdAt >= batch.createdAt; + }; + const actedOnCurrentVersion = (card: Card) => (lastAction.get(card.id) ?? -1) > (lastWrite.get(card.id) ?? -1); + const hiddenReason = (card: Card, run: SourceRun, forReview: boolean): string | undefined => { const recordedAt = run.completedAt ?? ""; - if (!card.sourceRunIds?.includes(run.id) || card.updatedAt < recordedAt) return false; - if (card.status === "to_review_new" || card.status === "to_review_updated") return card.readyForPass <= feed.config.currentPass; - if (card.status === "done") return (card.completedAt ?? "") >= recordedAt; - return workItems.some((work) => work.cardId === card.id && work.createdAt >= recordedAt); + if (!card.sourceRunIds?.includes(run.id)) return `card ${card.id} does not list run ${run.id} in sourceRunIds`; + if (card.updatedAt < recordedAt) return `card ${card.id} was last written before this run was recorded`; + if (forReview && card.routineActionGroupId) return `card ${card.id} belongs to routine action group ${card.routineActionGroupId}, so it is not individually reviewable; a review judgment needs its own card`; + if (!forReview && card.routineActionGroupId) { + const group = feed.routineActions.find((candidate) => candidate.id === card.routineActionGroupId); + if (group && !proposedForThisSweep(group)) return `card ${card.id} belongs to routine action group ${group.id}, which was proposed before this sweep; propose a group for this sweep or resurface the card`; + } + if (card.sweep?.hidden) return `card ${card.id} is hidden by sweep feedback ${card.sweep.feedbackId}; upsert it again with status to_review_updated if it must be presented`; + if (card.status === "to_review_new" || card.status === "to_review_updated") { + return card.readyForPass <= currentPass ? undefined : `card ${card.id} is deferred to pass ${card.readyForPass} (current pass ${currentPass})`; + } + if (card.status === "done") { + if ((card.completedAt ?? "") < recordedAt) return `card ${card.id} was dismissed or completed before this run and has not been resurfaced (upsert it with status to_review_updated)`; + return actedOnCurrentVersion(card) ? undefined : `card ${card.id} was updated after the user dismissed or completed it, so its current content was never reviewed; upsert it with status to_review_updated`; + } + if (!workItems.some((work) => work.cardId === card.id && work.createdAt >= recordedAt)) return `card ${card.id} is ${card.status} from before this run with no new approval since`; + return actedOnCurrentVersion(card) ? undefined : `card ${card.id} was updated after it was approved, so its current content was never reviewed; upsert it with status to_review_updated`; }; - let routineNeeded = 0; + type Judged = { run: SourceRun; gap: SweepPresentationGap; review: boolean; cardId?: string }; + const judged: Judged[] = []; + const runReports = new Map(); for (const run of runs) { - const decisions = run.judgments.flatMap((judgment) => isRecord(judgment) && typeof judgment.decision === "string" ? [judgment.decision] : []); - const reviews = decisions.filter((decision) => REVIEW_DECISIONS.has(decision)).length; - const routines = decisions.filter((decision) => decision === ROUTINE_DECISION).length; - const presented = cards.filter((card) => presentsRun(card, run)).length; - if (presented < reviews) { - throw new Error( - `Source run ${run.id} (${sourceDisplayName(run.sourceId)}) has ${reviews} review ${reviews === 1 ? "judgment" : "judgments"} but only ${presented} ${presented === 1 ? "card presents" : "cards present"} it. Upsert one card per review judgment with sourceRunIds including ${run.id}, in a review status for the current pass, before completing this work; Tend holds the source checkpoint until then.`, - ); + runReports.set(run.id, { runId: run.id, sourceId: run.sourceId, checkpointHeld: run.pendingCheckpoint !== undefined, judgments: run.judgments.length, needingPresentation: 0, presented: 0 }); + run.judgments.forEach((judgment, index) => { + if (!isRecord(judgment) || typeof judgment.decision !== "string") return; + const review = REVIEW_DECISIONS.has(judgment.decision); + if (!review && judgment.decision !== ROUTINE_DECISION) return; + runReports.get(run.id)!.needingPresentation += 1; + const cardId = typeof judgment.cardId === "string" ? judgment.cardId : undefined; + judged.push({ run, gap: { runId: run.id, sourceId: run.sourceId, judgment: index + 1, decision: judgment.decision, ...(cardId ? { cardId } : {}), reason: "" }, review, cardId }); + }); + } + const missing: SweepPresentationGap[] = []; + // Every card a judgment names is reserved for exact matching across the whole batch, whether or not it + // presents its run yet, so the report lists all gaps at once and a named card never silently satisfies an + // unnamed judgment elsewhere. Sharing a card is explicit: every judgment it presents names it. + const reserved = new Set(judged.flatMap((item) => item.cardId ? [item.cardId] : [])); + // Cards that presented a judgment (named or counted); a routine group item backed by one of them adds no capacity. + const presenting = new Set(); + for (const item of judged) { + if (item.cardId === undefined) continue; + const card = cardsById.get(item.cardId); + const reason = card ? hiddenReason(card, item.run, item.review) : `no card with id ${item.cardId} exists`; + if (reason) missing.push({ ...item.gap, reason }); + else { + runReports.get(item.run.id)!.presented += 1; + presenting.add(item.cardId); } - routineNeeded += Math.max(0, routines - (presented - reviews)); } - if (routineNeeded > 0) { - const earliestRecordedAt = runs.map((run) => run.completedAt ?? "").sort()[0] ?? ""; - const routineItems = feed.routineActions - .filter((group) => group.status !== "stale" && group.status !== "failed" && group.createdAt >= earliestRecordedAt) - .reduce((total, group) => total + group.items.length, 0); - if (routineItems < routineNeeded) { - throw new Error( - `This sweep has ${routineNeeded} routine_action ${routineNeeded === 1 ? "judgment" : "judgments"} without a card, but routine action groups proposed since it was recorded hold only ${routineItems} ${routineItems === 1 ? "item" : "items"}. Propose the routine action group or upsert cards for them before completing this work; Tend holds the source checkpoints until then.`, - ); + // Judgments without a cardId are matched by count, in batch order, over the run's presenting cards that no + // judgment reserved and no earlier match already used: a card counts once per batch unless judgments name it. + const routineWithoutCard: Judged[] = []; + for (const run of runs) { + const unnamed = judged.filter((item) => item.run === run && item.cardId === undefined); + const reviews = unnamed.filter((item) => item.review); + const routines = unnamed.filter((item) => !item.review); + const unreserved = cards.filter((card) => !reserved.has(card.id) && !presenting.has(card.id)); + const reviewCards = unreserved.filter((card) => hiddenReason(card, run, true) === undefined); + const routineOnlyCards = unreserved.filter((card) => hiddenReason(card, run, false) === undefined && hiddenReason(card, run, true) !== undefined); + const reviewPool = reviewCards.length; + const reviewCovered = Math.min(reviewPool, reviews.length); + const routineCandidates = [...reviewCards.slice(reviewCovered), ...routineOnlyCards]; + const routineCovered = Math.min(routineCandidates.length, routines.length); + for (const card of [...reviewCards.slice(0, reviewCovered), ...routineCandidates.slice(0, routineCovered)]) presenting.add(card.id); + runReports.get(run.id)!.presented += reviewCovered + routineCovered; + for (const item of reviews.slice(reviewCovered)) { + missing.push({ ...item.gap, reason: `no unclaimed card presents this review judgment (${reviewPool} available for ${reviews.length} review ${reviews.length === 1 ? "judgment" : "judgments"} without a cardId); add a cardId or upsert another card with sourceRunIds including ${run.id}` }); + } + routineWithoutCard.push(...routines.slice(routineCovered)); + } + // routine_action judgments without a card may be covered by items of routine action groups proposed since the + // sweep. Group items do not identify judgments, so this coverage is aggregate: when it falls short, every such + // judgment is listed rather than an invented subset. + // A group counts when it was proposed for this sweep, which its createdAt records: recording the batch marks + // earlier proposals stale, so a proposal for this sweep is created at or after the batch, and re-proposing a + // stale group under a reused id dates it from the re-proposal. Keying on the batch rather than the earliest + // run keeps a group created and approved between two runs from covering the later run's judgments. Status + // alone is not enough (an older approved group returns to `proposed` when its work is cancelled) and updatedAt + // is not used, since approval or completion of an older group refreshes it without presenting anything new. + const routineGroupItems = feed.routineActions + .filter((group) => group.status !== "stale" && group.status !== "failed" && proposedForThisSweep(group)) + .reduce((total, group) => total + group.items.filter((item) => !item.cardId || !presenting.has(item.cardId)).length, 0); + const routineCoveredByGroups = Math.min(routineGroupItems, routineWithoutCard.length); + if (routineGroupItems < routineWithoutCard.length) { + for (const item of routineWithoutCard) { + missing.push({ ...item.gap, reason: `no card presents this routine_action judgment, and the ${routineWithoutCard.length} such ${routineWithoutCard.length === 1 ? "judgment" : "judgments"} in this sweep ${routineWithoutCard.length === 1 ? "exceeds" : "exceed"} the ${routineGroupItems} routine action group ${routineGroupItems === 1 ? "item" : "items"} proposed since it was recorded; add a cardId, upsert a card, or propose the group` }); } } + const runOrder = new Map(runs.map((run, index) => [run.id, index])); + missing.sort((a, b) => (runOrder.get(a.runId)! - runOrder.get(b.runId)!) || (a.judgment - b.judgment)); + const reports = [...runReports.values()]; + const held = reports.filter((report) => report.checkpointHeld).length; + const needing = reports.reduce((total, report) => total + report.needingPresentation, 0); + const workId = batch.triggerWorkId ?? null; + // A run recorded for this work after the batch is neither presented nor superseded; the batch must be + // recorded again so its judgments are checked and its checkpoint commits with the others. + const unbatchedRuns = workId + ? (await this.store.listRuns(feedId)) + .filter((run) => run.triggerWorkId === workId && run.pendingCheckpoint !== undefined && !batch.sourceRunIds.includes(run.id) + && !runs.some((included) => included.sourceId === run.sourceId && (included.checkpointSequence ?? 0) > (run.checkpointSequence ?? 0))) + .map((run) => ({ runId: run.id, sourceId: run.sourceId })) + : []; + const presentationReady = missing.length === 0 && unbatchedRuns.length === 0; + const status = held || unbatchedRuns.length ? "pending" : "committed"; + const workStatus = workId ? workItems.find((work) => work.id === workId)?.status ?? null : null; + const completable = workStatus === "working" || workStatus === "queued"; + const ready = presentationReady && (status === "committed" || completable); + const describe = (gap: SweepPresentationGap) => `run ${gap.runId} judgment ${gap.judgment} (${gap.decision}${gap.cardId ? `, cardId ${gap.cardId}` : ""}): ${gap.reason}`; + const unbatched = unbatchedRuns.length + ? `Source ${unbatchedRuns.length === 1 ? "run" : "runs"} ${unbatchedRuns.map((run) => `${run.runId} (${sourceDisplayName(run.sourceId)})`).join(", ")} recorded for this work after batch ${batch.id} ${unbatchedRuns.length === 1 ? "is" : "are"} not in it; record the batch again including ${unbatchedRuns.length === 1 ? "it" : "them"} (or a newer run for the same source). ` + : ""; + const gaps = `${unbatched}${missing.length ? `${missing.length} of ${needing} judgments are not presented yet: ${missing.slice(0, 3).map(describe).join("; ")}${missing.length > 3 ? `; and ${missing.length - 3} more` : ""}.` : ""}`.trim(); + const checkpoints = `${held} held checkpoint${held === 1 ? "" : "s"}`; + let summary: string; + if (status === "committed") { + summary = `Batch ${batch.id} is committed; ${presentationReady ? `all ${needing} judgments that need presentation are presented.` : `${missing.length} of ${needing} judgments have no presenting card (informational).`}`; + } else if (!completable) { + // The owner can never complete, so say that first: presenting more cards will not commit these checkpoints. + summary = `Recollection work ${workId ?? "(none)"} is ${workStatus ?? "missing"} and cannot complete; request a new recollection to re-read from the ${checkpoints}. ${presentationReady ? `All ${needing} judgments that need presentation are presented.` : gaps}`; + } else if (!presentationReady) { + summary = gaps; + } else if (workStatus === "queued") { + summary = `All ${needing} judgments that need presentation are presented; claim work ${workId} again, then work:complete will commit the ${checkpoints}.`; + } else { + summary = `All ${needing} judgments that need presentation are presented; work:complete will commit the ${checkpoints}.`; + } + return { status, currentBatchId: batch.id, workId, workStatus, ready, runs: reports, missing, unbatchedRuns, routineGroupItems, routineCoveredByGroups, summary }; + } + + // Recollection work completes only once every judgment that needs presentation has it; that is when the + // checkpoints held on the batch's runs become the sources' live checkpoints, inside the same transaction. + // The whole batch is validated before anything is written so a refusal leaves both SQLite and the file mirrors untouched. + private async commitHeldSourceCheckpoints(feedId: string, batch: SweepBatch): Promise { + const report = await this.presentationReport(feedId, batch); + if (!report.ready) { + throw new Error(`${report.summary} Tend holds the source checkpoints until then; run \`tend cli sweep:status --feed ${feedId}\` for the full list.`); + } + const runs = await Promise.all(batch.sourceRunIds.map((runId) => this.store.readRun(feedId, runId))); // The newest run per source (by recording sequence, then recorded time) owns that source's checkpoint; // earlier same-source runs in the batch are superseded. const recordingOrder = (a: SourceRun, b: SourceRun) => (a.checkpointSequence ?? 0) - (b.checkpointSequence ?? 0) || (a.completedAt ?? "").localeCompare(b.completedAt ?? ""); @@ -3810,6 +3989,17 @@ export class AttentionDomain { const committedAt = isoNow(); const committed: string[] = []; const skipped: Array<{ runId: string; reason: string }> = []; + // Runs recorded for this work but left out of the batch were superseded by a newer run for their source + // (record-batch refuses any other omission); settle them so nothing stays held after the work completes. + if (batch.triggerWorkId) { + for (const run of await this.store.listRuns(feedId)) { + if (run.triggerWorkId !== batch.triggerWorkId || run.pendingCheckpoint === undefined || batch.sourceRunIds.includes(run.id)) continue; + skipped.push({ runId: run.id, reason: "superseded by a newer run for the same source, which the batch included" }); + const settled: SourceRun = { ...run, checkpointCommittedAt: committedAt }; + delete settled.pendingCheckpoint; + await this.store.writeRun(settled); + } + } for (const run of runs) { if (run.pendingCheckpoint === undefined) continue; if (newestBySource.get(run.sourceId) === run) { diff --git a/server/operator.ts b/server/operator.ts index 1957e1c..e7a34dd 100644 --- a/server/operator.ts +++ b/server/operator.ts @@ -1,4 +1,4 @@ -import type { Card, CardBlock, FeedConfig, ProposedAction, RoutineActionGroup, SweepFeedbackTrace, WorkClaimResult, WorkClaimedByReport, WorkItem, WorkItemView } from "../shared/types"; +import type { Card, CardBlock, FeedConfig, ProposedAction, RoutineActionGroup, SweepFeedbackTrace, SweepPresentationStatus, WorkClaimResult, WorkClaimedByReport, WorkItem, WorkItemView } from "../shared/types"; import { actionEmailRecipients } from "../shared/emailRecipients"; import { actionDigest, cleanupDigest, configuredApprovalAction, routineActionDigest } from "./workflow/approvals"; @@ -26,6 +26,8 @@ export interface ClaimedWorkOutput extends Omit; routineActionGroup?: RoutineActionGroup; sweepFeedback?: Pick; + sweepPresentation?: SweepPresentationStatus; } export interface UserAuthorizationReceipt { @@ -256,7 +259,16 @@ export function formatWorkClaimOutput(feedId: string, work: WorkClaimResult, con } if (work.intent === "recollect_sources") { - operatorGuidance.requiredWriteBack = "Record one or more source runs with `source:record-run --work `, then create a sweep batch with `sweep:record-batch --work `, then upsert one card (with sourceRunIds) per review judgment and present routine_action judgments as cards or a proposed routine action group before `work:complete`. Checkpoints recorded with --work advance only when completion succeeds; completion is refused until every judgment is presented."; + operatorGuidance.requiredWriteBack = "Record one or more source runs with `source:record-run --work `, giving every review or routine_action judgment a stable `cardId` (for example `gmail-`) that you will reuse as the card id in `card:upsert`; then `sweep:record-batch --work `; then upsert one card per judgment with sourceRunIds including the run. Run `sweep:status --feed ` to see exactly what is still missing; `work:complete` is refused until it reports ready, and the source checkpoints advance only then. Judgments without a cardId are matched by count instead."; + const pending = context.sweepPresentation; + if (pending?.status === "pending" && pending.workId === work.id) { + operatorGuidance.completionPrerequisite = pending.unbatchedRuns.length + ? `Batch ${pending.currentBatchId} was recorded by this work, but ${pending.unbatchedRuns.map((run) => run.runId).join(", ")} recorded afterwards ${pending.unbatchedRuns.length === 1 ? "is" : "are"} not in it. Record the batch again with sweep:record-batch --work ${work.id} including ${pending.unbatchedRuns.length === 1 ? "it" : "them"} (or a newer run for the same source), then rerun sweep:status --feed ${feedId} for the fresh list of judgments to present, then work:complete.` + : pending.ready + ? `Batch ${pending.currentBatchId} recorded by this work is fully presented; finish with work:complete.` + : `Batch ${pending.currentBatchId} was already recorded by this work; present the judgments listed in pendingPresentation.missing, rerun sweep:status --feed ${feedId}, then work:complete. If a listed card cannot be updated (for example an immutable reading card), record a corrected source run that names a different cardId and record the batch again including it.`; + operatorGuidance.pendingPresentation = pending; + } operatorGuidance.sourceRunRule = feedId === "inbox" ? "For a full Gmail sweep, first paginate gmail_search_email_ids(query='', label_ids=['INBOX']). Treat that message-ID manifest as authoritative, direct-read every ID, and record an inboxEnumeration.messages entry mapping each messageId to its threadId. Its readThreadIds and carriedForwardThreadIds must then classify every resulting thread exactly once. gmail_search_emails results may enrich the run but cannot define the Inbox universe. Source recollection must complete with a new sweep batch recorded for this exact work item." : "Source recollection work must complete with a new sweep batch recorded for this exact work item."; diff --git a/server/templates.ts b/server/templates.ts index 3f7db03..6a3fa3c 100644 --- a/server/templates.ts +++ b/server/templates.ts @@ -22,6 +22,9 @@ Before composing full cards, choose a disposition for each source item: \`review deserve individual attention, \`routine_action\` for a conservative batch with the same obvious cleanup, or \`suppress\` for items that should stay out of the sweep. Routine actions remain proposed until the user approves the exact visible group. Never hide ambiguity inside a routine-action batch. +Give every \`review\` or \`routine_action\` judgment a stable \`cardId\` (for example \`gmail-\`) and +reuse that exact id in \`card:upsert\`; Tend verifies each judged item was presented before the source +checkpoint advances. `; export const COMPOSE_CARD_PROMPT = `# Compose card prompt diff --git a/server/version.ts b/server/version.ts index 2ccf940..f6edfff 100644 --- a/server/version.ts +++ b/server/version.ts @@ -2,7 +2,7 @@ import packageJson from "../package.json"; export const APP_NAME = packageJson.name; export const APP_VERSION = packageJson.version; -export const CLI_CONTRACT_VERSION = "0.8"; +export const CLI_CONTRACT_VERSION = "0.9"; export function versionInfo() { return { diff --git a/shared/types.ts b/shared/types.ts index c1f180f..b037c45 100644 --- a/shared/types.ts +++ b/shared/types.ts @@ -653,6 +653,45 @@ export interface SourceRun { checkpointCommittedAt?: string; } +/** One judgment of the current sweep that is not presented yet, with the exact reason. */ +export interface SweepPresentationGap { + runId: string; + sourceId: string; + /** 1-based position in the run's judgments array. */ + judgment: number; + decision: string; + cardId?: string; + reason: string; +} + +export interface SweepPresentationRun { + runId: string; + sourceId: string; + checkpointHeld: boolean; + judgments: number; + needingPresentation: number; + /** Judgments presented by cards (exact matches plus counted cards). Routine judgments covered by group items are reported batch-wide in `routineCoveredByGroups`. */ + presented: number; +} + +/** What `sweep:status` reports and what `work:complete` checks for recollection work. */ +export interface SweepPresentationStatus { + status: "idle" | "committed" | "pending"; + currentBatchId: string | null; + workId: string | null; + workStatus: WorkStatus | null; + ready: boolean; + runs: SweepPresentationRun[]; + missing: SweepPresentationGap[]; + /** Held runs recorded for the batch's work that the batch neither includes nor supersedes; the batch must be recorded again. */ + unbatchedRuns: Array<{ runId: string; sourceId: string }>; + /** Items of routine action groups proposed since the sweep, available to cover routine_action judgments without a card. */ + routineGroupItems: number; + /** routine_action judgments without a card that those group items cover (aggregate, not attributable to a run). */ + routineCoveredByGroups: number; + summary: string; +} + export interface AppFeedback { id: string; feedId: FeedId; diff --git a/test/cli-contract.test.ts b/test/cli-contract.test.ts index 1ed4a8b..641dddb 100644 --- a/test/cli-contract.test.ts +++ b/test/cli-contract.test.ts @@ -29,6 +29,7 @@ describe("CLI contract", () => { expect(commandNames).toContain("work:complete"); expect(commandNames).toContain("work:reconcile-approved"); expect(commandNames).toContain("source:record-run"); + expect(commandNames).toContain("sweep:status"); expect(commandNames).toContain("card:upsert"); expect(commandNames).toContain("card:dismiss"); expect(commandNames).toContain("card:cleanup-source"); @@ -43,7 +44,7 @@ describe("CLI contract", () => { }); test("exposes native readers and exact-version feedback without edition commands", () => { - expect(CLI_CONTRACT_VERSION).toBe("0.8"); + expect(CLI_CONTRACT_VERSION).toBe("0.9"); expect(CLI_COMMANDS).toContain("readers:run --feed --run --packet-file --readers-file [--prompt-sha256 ]"); expect(CLI_COMMANDS).toContain("readers:status --feed --run "); expect(CLI_COMMANDS).toContain("readers:output --feed --run --reader "); @@ -218,6 +219,14 @@ describe("CLI contract", () => { const runId = JSON.parse(recorded.stdout) as string; const store = new AttentionStore(path.join(home, "data")); expect(await store.readRun("company-attention", runId)).toMatchObject({ snapshots: 1, judgments: [] }); + const idle = await run(["sweep:status", "--feed", "company-attention"]); + expect(idle.code).toBe(0); + expect(JSON.parse(idle.stdout)).toMatchObject({ status: "idle", ready: true, missing: [] }); + const batch = await run(["sweep:record-batch", "--feed", "company-attention", "--runs", JSON.stringify([runId])]); + expect(batch.code).toBe(0); + const committed = await run(["sweep:status", "--feed", "company-attention"]); + expect(committed.code).toBe(0); + expect(JSON.parse(committed.stdout)).toMatchObject({ status: "committed", currentBatchId: JSON.parse(batch.stdout), ready: true, missing: [], runs: [{ runId, checkpointHeld: false, judgments: 0, needingPresentation: 0, presented: 0 }] }); const saved = JSON.parse(await readFile(store.feedPath("company-attention", "raw", runId, "company-attention", "snapshot-1.json"), "utf8")); expect(saved).toEqual(snapshots[0]); const missing = await run(["readers:run", "--feed", "company-attention", "--run", runId, "--packet-file", "unused.txt"]); diff --git a/test/sweep-checkpoints.test.ts b/test/sweep-checkpoints.test.ts index 414b369..c096ec6 100644 --- a/test/sweep-checkpoints.test.ts +++ b/test/sweep-checkpoints.test.ts @@ -4,6 +4,7 @@ import os from "node:os"; import path from "node:path"; import { AttentionDomain } from "../server/domain"; import { createLocalRuntime } from "../server/runtime"; +import { formatWorkClaimOutput } from "../server/operator"; import type { WorkItem } from "../shared/types"; const FEED = "company-attention"; @@ -43,7 +44,7 @@ test("a recollection checkpoint advances only when the work completes with cards expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Collected." })) - .rejects.toThrow(`Source run ${run} (Company Attention) has 1 review judgment but only 0 cards present it`); + .rejects.toThrow(`1 of 1 judgments are not presented yet: run ${run} judgment 1 (review): no unclaimed card presents this review judgment (0 available for 1 review judgment without a cardId)`); expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); expect(await mirrorCheckpoint()).toEqual(before); expect((await store.readWork(FEED, work.id)).status).toBe("working"); @@ -143,7 +144,7 @@ test("a refused completion writes nothing, even when an earlier run in the batch const missing = await domain.recordSourceRun(FEED, second.id, [{ note: "Not presented." }], [{ decision: "review" }], { cursor: "missing" }, work.id); await domain.recordSweepBatch(FEED, [presented, missing], work.id); await domain.upsertCard(FEED, { id: "presented-card", title: "Presented", why: "Has a card.", blocks: [], sourceRunIds: [presented] }); - await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).rejects.toThrow(`Source run ${missing}`); + await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).rejects.toThrow(`1 of 2 judgments are not presented yet: run ${missing} judgment 1 (review)`); expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); expect(await mirrorCheckpoint()).toEqual(beforeMirror); expect((await store.readRun(FEED, presented)).pendingCheckpoint).toEqual({ cursor: "presented" }); @@ -163,10 +164,10 @@ test("every review judgment needs its own card, and routine_action judgments nee await domain.recordSweepBatch(FEED, [run], work.id); await domain.upsertCard(FEED, { id: "only-one", title: "One", why: "Covers one review judgment.", blocks: [], sourceRunIds: [run] }); await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })) - .rejects.toThrow(`has 2 review judgments but only 1 card presents it`); + .rejects.toThrow("2 of 3 judgments are not presented yet: run " + run + " judgment 2 (review): no unclaimed card presents this review judgment (1 available for 2 review judgments without a cardId)"); await domain.upsertCard(FEED, { id: "the-other", title: "Two", why: "Covers the other review judgment.", blocks: [], sourceRunIds: [run] }); await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })) - .rejects.toThrow(`This sweep has 1 routine_action judgment without a card, but routine action groups proposed since it was recorded hold only 0 items`); + .rejects.toThrow("judgment 3 (routine_action): no card presents this routine_action judgment, and the 1 such judgment in this sweep exceeds the 0 routine action group items proposed since it was recorded"); expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); await domain.upsertRoutineActionGroup(FEED, { id: "routine-fixture", label: "Archive newsletters", summary: "Three newsletters with the same obvious cleanup.", @@ -221,7 +222,7 @@ test("a dismissed card that merely gains the run id does not count as presented" await domain.upsertCard(FEED, { id: "old-thread", title: "Old thread", why: "A new message arrived.", blocks: [], sourceRunIds: [run] }); expect((await store.readCard(FEED, "old-thread")).status).toBe("done"); await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })) - .rejects.toThrow("has 1 review judgment but only 0 cards present it"); + .rejects.toThrow("1 of 1 judgments are not presented yet: run " + run + " judgment 1 (review): no unclaimed card presents this review judgment"); expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); await domain.upsertCard(FEED, { id: "old-thread", title: "Old thread", why: "A new message arrived.", blocks: [], sourceRunIds: [run], status: "to_review_updated" }); expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); @@ -246,7 +247,7 @@ test("cards deferred to a later pass or merely re-tagged while active do not cou await domain.upsertCard(FEED, { id: "already-queued", title: "Queued earlier", why: "Approved before the sweep.", blocks: [], sourceRunIds: [run] }); expect((await store.readCard(FEED, "already-queued")).status).toBe("queued"); await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })) - .rejects.toThrow("has 2 review judgments but only 0 cards present it"); + .rejects.toThrow("2 of 2 judgments are not presented yet"); expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); await domain.upsertCard(FEED, { id: "deferred", title: "Deferred", why: "Now visible.", blocks: [], sourceRunIds: [run], readyForPass: currentPass }); await domain.upsertCard(FEED, { id: "already-queued", title: "Queued earlier", why: "Back in review with new evidence.", blocks: [], sourceRunIds: [run], status: "to_review_updated" }); @@ -271,7 +272,7 @@ test("a routine action group proposed before the sweep does not present its rout const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }, { b: 2 }], [{ decision: "routine_action" }, { decision: "routine_action" }], { cursor: "routine" }, work.id); await domain.recordSweepBatch(FEED, [run], work.id); // supersedes the earlier proposed group await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })) - .rejects.toThrow("This sweep has 2 routine_action judgments without a card, but routine action groups proposed since it was recorded hold only 0 items"); + .rejects.toThrow("2 of 2 judgments are not presented yet: run " + run + " judgment 1 (routine_action): no card presents this routine_action judgment, and the 2 such judgments in this sweep exceed the 0 routine action group items proposed since it was recorded"); expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); await domain.upsertRoutineActionGroup(FEED, { id: "fresh-routine", label: "New batch", summary: "Proposed for this sweep.", @@ -279,12 +280,15 @@ test("a routine action group proposed before the sweep does not present its rout items: [{ id: "new-1", title: "Digest", reason: "Routine." }], }); await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })) - .rejects.toThrow("hold only 1 item"); + .rejects.toThrow("2 of 2 judgments are not presented yet: run " + run + " judgment 1 (routine_action): no card presents this routine_action judgment, and the 2 such judgments in this sweep exceed the 1 routine action group item proposed since it was recorded"); await domain.upsertRoutineActionGroup(FEED, { id: "fresh-routine", label: "New batch", summary: "Proposed for this sweep.", proposedAction: { label: "Archive", instruction: "Archive the listed items." }, items: [{ id: "new-1", title: "Digest", reason: "Routine." }, { id: "new-2", title: "Other digest", reason: "Routine." }], }); + const covered = await domain.sweepPresentationStatus(FEED); + expect(covered).toMatchObject({ ready: true, routineGroupItems: 2, routineCoveredByGroups: 2 }); + expect(covered.runs[0]).toMatchObject({ needingPresentation: 2, presented: 0 }); expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual({ cursor: "routine" }); } finally { @@ -293,6 +297,595 @@ test("a routine action group proposed before the sweep does not present its rout } }); +test("judgments that name a cardId are matched against that exact card, with precise reasons", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const before = await store.readSourceCheckpoint(FEED, SOURCE); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ status: "idle", ready: true, missing: [] }); + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ thread: "a" }, { thread: "b" }, { thread: "c" }], + [{ decision: "review", cardId: "thread-a" }, { decision: "review", cardId: "thread-b" }, { decision: "suppress", cardId: "ignored" }], { cursor: "exact" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + + let status = await domain.sweepPresentationStatus(FEED); + expect(status).toMatchObject({ status: "pending", currentBatchId: expect.any(String), workId: work.id, workStatus: "working", ready: false }); + expect(status.missing.map((gap) => [gap.judgment, gap.cardId, gap.reason])).toEqual([ + [1, "thread-a", "no card with id thread-a exists"], + [2, "thread-b", "no card with id thread-b exists"], + ]); + await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })) + .rejects.toThrow("2 of 2 judgments are not presented yet: run " + run + " judgment 1 (review, cardId thread-a): no card with id thread-a exists"); + + // The right card id but no provenance, then a card that was dismissed before this run. + await domain.upsertCard(FEED, { id: "thread-a", title: "A", why: "Forgot sourceRunIds.", blocks: [] }); + await domain.upsertCard(FEED, { id: "thread-b", title: "B", why: "Old.", blocks: [], sourceRunIds: [run], status: "done", completionDisposition: "dismissed", completedAt: "2020-01-01T00:00:00.000Z" }); + status = await domain.sweepPresentationStatus(FEED); + expect(status.missing.map((gap) => gap.reason)).toEqual([ + `card thread-a does not list run ${run} in sourceRunIds`, + "card thread-b was dismissed or completed before this run and has not been resurfaced (upsert it with status to_review_updated)", + ]); + expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual(before); + + await domain.upsertCard(FEED, { id: "thread-a", title: "A", why: "Now with provenance.", blocks: [], sourceRunIds: [run] }); + await domain.upsertCard(FEED, { id: "thread-b", title: "B", why: "Resurfaced.", blocks: [], sourceRunIds: [run], status: "to_review_updated" }); + status = await domain.sweepPresentationStatus(FEED); + expect(status.ready).toBe(true); + expect(status.runs[0]).toMatchObject({ runId: run, checkpointHeld: true, judgments: 3, needingPresentation: 2, presented: 2 }); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual({ cursor: "exact" }); + expect((await domain.sweepPresentationStatus(FEED)).status).toBe("committed"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a card claimed by an exact match does not also count for a judgment without a cardId", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }, { b: 2 }], [{ decision: "review", cardId: "named" }, { decision: "review" }], { cursor: "mixed" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { id: "named", title: "Named", why: "Exact match.", blocks: [], sourceRunIds: [run] }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.missing).toHaveLength(1); + expect(status.missing[0]).toMatchObject({ judgment: 2, decision: "review" }); + expect(status.missing[0].cardId).toBeUndefined(); + expect(status.missing[0].reason).toContain("0 available for 1 review judgment without a cardId"); + await domain.upsertCard(FEED, { id: "anonymous", title: "Other", why: "Counted.", blocks: [], sourceRunIds: [run] }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual({ cursor: "mixed" }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +for (const order of ["named-first", "unnamed-first"] as const) { + test(`a card named by one run's judgment is not counted for another run's unnamed judgment (${order})`, async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const second = await domain.addSourceFromBrief(FEED, "Read the dispute ledger."); + const named = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "shared" }], { cursor: "named" }, work.id); + const unnamed = await domain.recordSourceRun(FEED, second.id, [{ b: 2 }], [{ decision: "review" }], { cursor: "unnamed" }, work.id); + await domain.recordSweepBatch(FEED, order === "named-first" ? [named, unnamed] : [unnamed, named], work.id); + await domain.upsertCard(FEED, { id: "shared", title: "Shared", why: "Lists both runs but is named by only one judgment.", blocks: [], sourceRunIds: [named, unnamed] }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.ready).toBe(false); + expect(status.missing.map((gap) => [gap.runId, gap.judgment, gap.cardId])).toEqual([[unnamed, 1, undefined]]); + await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).rejects.toThrow("1 of 2 judgments are not presented yet"); + // Naming the same card from the second judgment makes the sharing explicit. + await domain.upsertCard(FEED, { id: "shared", title: "Shared", why: "Same card, now named by both.", blocks: [], sourceRunIds: [named, unnamed] }); + const rerecorded = await domain.recordSourceRun(FEED, second.id, [{ b: 2 }], [{ decision: "review", cardId: "shared" }], { cursor: "unnamed" }, work.id); + await domain.recordSweepBatch(FEED, [named, rerecorded], work.id); + await domain.upsertCard(FEED, { id: "shared", title: "Shared", why: "Same card, now named by both.", blocks: [], sourceRunIds: [named, rerecorded] }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual({ cursor: "named" }); + expect(await store.readSourceCheckpoint(FEED, second.id)).toEqual({ cursor: "unnamed" }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } + }); +} + +for (const order of ["named-first", "unnamed-first"] as const) { + test(`a named card that does not yet present its run is still reserved, and gaps follow batch order (${order})`, async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const second = await domain.addSourceFromBrief(FEED, "Read the dispute ledger."); + const named = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "shared" }], { cursor: "named" }, work.id); + const unnamed = await domain.recordSourceRun(FEED, second.id, [{ b: 2 }], [{ decision: "review" }], { cursor: "unnamed" }, work.id); + const batchOrder = order === "named-first" ? [named, unnamed] : [unnamed, named]; + await domain.recordSweepBatch(FEED, batchOrder, work.id); + // The card exists but lists only the unnamed run: it must not count for that run's unnamed judgment either. + await domain.upsertCard(FEED, { id: "shared", title: "Shared", why: "Wrong provenance.", blocks: [], sourceRunIds: [unnamed] }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.missing.map((gap) => gap.runId)).toEqual(batchOrder); + expect(status.missing.find((gap) => gap.runId === named)).toMatchObject({ judgment: 1, cardId: "shared", reason: `card shared does not list run ${named} in sourceRunIds` }); + expect(status.missing.find((gap) => gap.runId === unnamed)?.cardId).toBeUndefined(); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } + }); +} + +test("readiness reflects the work's state, and the missing list follows batch order", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }, { b: 2 }, { c: 3 }], + [{ decision: "review" }, { decision: "review", cardId: "named" }, { decision: "routine_action" }], { cursor: "ordered" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + let status = await domain.sweepPresentationStatus(FEED); + expect(status.missing.map((gap) => gap.judgment)).toEqual([1, 2, 3]); + await domain.upsertCard(FEED, { id: "named", title: "Named", why: "Exact.", blocks: [], sourceRunIds: [run] }); + await domain.upsertCard(FEED, { id: "counted-1", title: "One", why: "Counted.", blocks: [], sourceRunIds: [run] }); + await domain.upsertCard(FEED, { id: "counted-2", title: "Two", why: "Counted.", blocks: [], sourceRunIds: [run] }); + status = await domain.sweepPresentationStatus(FEED); + expect(status).toMatchObject({ ready: true, workStatus: "working" }); + expect(status.summary).toContain("work:complete will commit the 1 held checkpoint"); + + await domain.releaseWork(FEED, work.id, work.capabilityToken); + status = await domain.sweepPresentationStatus(FEED); + expect(status).toMatchObject({ ready: true, workStatus: "queued" }); + expect(status.summary).toContain(`claim work ${work.id} again`); + + const reclaimed = await domain.claimWork(FEED, THREAD) as WorkItem; + await domain.failWork(FEED, work.id, reclaimed.capabilityToken, "Connector died."); + status = await domain.sweepPresentationStatus(FEED); + expect(status).toMatchObject({ ready: false, workStatus: "failed", missing: [] }); + expect(status.summary.startsWith(`Recollection work ${work.id} is failed and cannot complete`)).toBe(true); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a failed owner is reported before the gap list", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "never-made" }], { cursor: "failed" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.failWork(FEED, work.id, work.capabilityToken, "Died before presenting."); + const status = await domain.sweepPresentationStatus(FEED); + expect(status).toMatchObject({ status: "pending", ready: false, workStatus: "failed" }); + expect(status.missing).toHaveLength(1); + expect(status.summary.startsWith(`Recollection work ${work.id} is failed and cannot complete; request a new recollection`)).toBe(true); + expect(status.summary).toContain("1 of 1 judgments are not presented yet"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a review card pulled into a proposed routine group no longer presents its judgment", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "x" }], { cursor: "grouped" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { id: "x", title: "X", why: "Reviewable.", blocks: [], sourceRunIds: [run] }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + await domain.upsertRoutineActionGroup(FEED, { + id: "group-with-x", label: "Batch", summary: "Contains the review card.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "item-x", cardId: "x", title: "X", reason: "Routine." }], + }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.ready).toBe(false); + expect(status.missing[0].reason).toBe("card x belongs to routine action group group-with-x, so it is not individually reviewable; a review judgment needs its own card"); + await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).rejects.toThrow("not individually reviewable"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a routine action group re-proposed under a reused id counts for the new sweep", async () => { + const { root, runtime, store, domain } = await setup(); + try { + await domain.upsertRoutineActionGroup(FEED, { + id: "weekly-digests", label: "Digests", summary: "Last sweep's group.", + proposedAction: { label: "Archive", instruction: "Archive the digests." }, + items: [{ id: "old", title: "Old digest", reason: "Routine." }], + }); + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "routine_action" }], { cursor: "reused" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); // supersedes the earlier proposal + expect((await store.readFeed(FEED)).routineActions.find((group) => group.id === "weekly-digests")?.status).toBe("stale"); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(false); + await domain.upsertRoutineActionGroup(FEED, { + id: "weekly-digests", label: "Digests", summary: "This sweep's group, same id.", + proposedAction: { label: "Archive", instruction: "Archive the digests." }, + items: [{ id: "new", title: "New digest", reason: "Routine." }], + }); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: true, routineGroupItems: 1, routineCoveredByGroups: 1 }); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a batch must include every held run recorded for its work unless a newer run for that source supersedes it", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const second = await domain.addSourceFromBrief(FEED, "Read the dispute ledger."); + const a = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "a-card" }], { cursor: "a" }, work.id); + const b = await domain.recordSourceRun(FEED, second.id, [{ b: 1 }], [{ decision: "suppress" }], { cursor: "b" }, work.id); + await expect(domain.recordSweepBatch(FEED, [b], work.id)).rejects.toThrow(`Source run ${a} (Company Attention) recorded for this recollection is missing from the batch`); + const aAgain = await domain.recordSourceRun(FEED, SOURCE, [{ a: 2 }], [{ decision: "suppress" }], { cursor: "a2" }, work.id); + await domain.recordSweepBatch(FEED, [b, aAgain], work.id); // the newer run for the same source supersedes the first + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual({ cursor: "a2" }); + const superseded = await store.readRun(FEED, a); + expect(superseded.pendingCheckpoint).toBeUndefined(); + expect(typeof superseded.checkpointCommittedAt).toBe("string"); + const event = (await store.readEvents(FEED)).find((item) => item.type === "sweep.checkpoints_committed"); + expect(event?.detail).toMatchObject({ skipped: expect.arrayContaining([{ runId: a, reason: "superseded by a newer run for the same source, which the batch included" }]) }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a card that already presents a judgment adds no routine group capacity", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }, { b: 2 }], [{ decision: "routine_action", cardId: "x" }, { decision: "routine_action" }], { cursor: "double" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { id: "x", title: "X", why: "Presents judgment 1.", blocks: [], sourceRunIds: [run] }); + await domain.upsertRoutineActionGroup(FEED, { + id: "one-item", label: "Batch", summary: "Only card x.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "item-x", cardId: "x", title: "X", reason: "Routine." }], + }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status).toMatchObject({ ready: false, routineGroupItems: 0, routineCoveredByGroups: 0 }); + expect(status.missing.map((gap) => gap.judgment)).toEqual([2]); + await domain.upsertRoutineActionGroup(FEED, { + id: "one-item", label: "Batch", summary: "Card x plus a card-less item for judgment 2.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "item-x", cardId: "x", title: "X", reason: "Routine." }, { id: "item-2", title: "Other", reason: "Routine." }], + }); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: true, routineGroupItems: 1, routineCoveredByGroups: 1 }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a run recorded after the batch must be batched before the work can complete", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const second = await domain.addSourceFromBrief(FEED, "Read the dispute ledger."); + const a = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "suppress" }], { cursor: "a" }, work.id); + await domain.recordSweepBatch(FEED, [a], work.id); + const b = await domain.recordSourceRun(FEED, second.id, [{ b: 1 }], [{ decision: "review", cardId: "b-card" }], { cursor: "b" }, work.id); + const status = await domain.sweepPresentationStatus(FEED); + expect(status).toMatchObject({ ready: false, status: "pending", unbatchedRuns: [{ runId: b, sourceId: second.id }] }); + expect(status.summary).toMatch(new RegExp(`^Source run ${b} \\(.+\\) recorded for this work after batch .+ is not in it; record the batch again`)); + await expect(domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).rejects.toThrow("record the batch again"); + expect((await store.readRun(FEED, b)).pendingCheckpoint).toEqual({ cursor: "b" }); + await domain.recordSweepBatch(FEED, [a, b], work.id); + await domain.upsertCard(FEED, { id: "b-card", title: "B", why: "Presented.", blocks: [], sourceRunIds: [b] }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual({ cursor: "a" }); + expect(await store.readSourceCheckpoint(FEED, second.id)).toEqual({ cursor: "b" }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a card counted for one run's unnamed judgment is not counted again for another run", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const second = await domain.addSourceFromBrief(FEED, "Read the dispute ledger."); + const a = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review" }], { cursor: "a" }, work.id); + const b = await domain.recordSourceRun(FEED, second.id, [{ b: 1 }], [{ decision: "review" }], { cursor: "b" }, work.id); + await domain.recordSweepBatch(FEED, [a, b], work.id); + await domain.upsertCard(FEED, { id: "both", title: "Both", why: "Lists both runs without being named by either judgment.", blocks: [], sourceRunIds: [a, b] }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.ready).toBe(false); + expect(status.missing.map((gap) => gap.runId)).toEqual([b]); + expect(status.runs.map((run) => run.presented)).toEqual([1, 0]); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a card dismissed between the run and its refresh does not present the refreshed content", async () => { + const { root, runtime, store, domain } = await setup(); + try { + await domain.upsertCard(FEED, { id: "thread-r", title: "Old content", why: "Seen before.", blocks: [] }); + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ thread: "thread-r" }], [{ decision: "review", cardId: "thread-r" }], { cursor: "race" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + // The user dismisses the old card after the run was recorded, then the agent refreshes it without resurfacing. + await domain.dismissCard(FEED, "thread-r"); + await domain.upsertCard(FEED, { id: "thread-r", title: "New content", why: "A new message arrived.", blocks: [], sourceRunIds: [run] }); + expect((await store.readCard(FEED, "thread-r")).status).toBe("done"); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.ready).toBe(false); + expect(status.missing[0].reason).toBe("card thread-r was updated after the user dismissed or completed it, so its current content was never reviewed; upsert it with status to_review_updated"); + await domain.upsertCard(FEED, { id: "thread-r", title: "New content", why: "A new message arrived.", blocks: [], sourceRunIds: [run], status: "to_review_updated" }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + // A dismissal of the presented content still counts. + await domain.dismissCard(FEED, "thread-r"); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a card refreshed after its approval does not present the refreshed content", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "approve-me" }], { cursor: "approval" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { + id: "approve-me", title: "Approve me", why: "Has an approvable action.", sourceRunIds: [run], + blocks: [{ id: "draft", type: "editable_text", value: "Draft.", editable: true }], + actions: [{ id: "send", label: "Send", behavior: "approve_action", instruction: "Send the draft.", artifactBlockId: "draft", externalMutation: true }], + }); + await domain.runCardAction(FEED, "approve-me", "send"); + expect((await store.readCard(FEED, "approve-me")).status).toBe("queued"); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); // approved after presentation + await domain.upsertCard(FEED, { + id: "approve-me", title: "Approve me", why: "Refreshed after approval.", sourceRunIds: [run], + blocks: [{ id: "draft", type: "editable_text", value: "Changed draft.", editable: true }], + actions: [{ id: "send", label: "Send", behavior: "approve_action", instruction: "Send the draft.", artifactBlockId: "draft", externalMutation: true }], + }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.ready).toBe(false); + expect(status.missing[0].reason).toBe("card approve-me was updated after it was approved, so its current content was never reviewed; upsert it with status to_review_updated"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a card the user queued an instruction or cleanup for counts as reviewed", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }, { b: 2 }], [{ decision: "review", cardId: "instructed" }, { decision: "review", cardId: "cleaned" }], { cursor: "dispositions" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { id: "instructed", title: "Instructed", why: "The user will queue an instruction.", blocks: [], sourceRunIds: [run] }); + await domain.upsertCard(FEED, { id: "cleaned", title: "Cleaned", why: "The user will queue source cleanup.", blocks: [], sourceRunIds: [run], actions: [{ id: "archive", label: "Archive", behavior: "default_cleanup" }] }); + await domain.queueInstruction(FEED, "instructed", "Draft a reply to this."); + await domain.queueSourceCleanup(FEED, "cleaned"); + expect((await store.readCard(FEED, "instructed")).status).toBe("queued"); + expect((await store.readCard(FEED, "cleaned")).status).toBe("queued"); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a group created and approved between two runs does not cover the later run's judgment", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const second = await domain.addSourceFromBrief(FEED, "Read the dispute ledger."); + const a = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "suppress" }], { cursor: "a" }, work.id); + await domain.upsertRoutineActionGroup(FEED, { + id: "between", label: "Between", summary: "Proposed and approved between the runs.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "i1", title: "One", reason: "Routine." }], + }); + await domain.approveRoutineActionGroup(FEED, "between"); + const b = await domain.recordSourceRun(FEED, second.id, [{ b: 1 }], [{ decision: "routine_action" }], { cursor: "b" }, work.id); + await domain.recordSweepBatch(FEED, [a, b], work.id); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: false, routineGroupItems: 0 }); + await domain.upsertRoutineActionGroup(FEED, { + id: "for-this-sweep", label: "Now", summary: "Proposed after the batch.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "i2", title: "Two", reason: "Routine." }], + }); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: true, routineGroupItems: 1 }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a group proposed just before the batch does not count even if their timestamps tie", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "routine_action" }], { cursor: "tie" }, work.id); + await domain.upsertRoutineActionGroup(FEED, { + id: "just-before", label: "Before", summary: "Proposed before the batch.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "i1", title: "One", reason: "Routine." }], + }); + await domain.approveRoutineActionGroup(FEED, "just-before"); // queued, so recording the batch does not stale it + const batchId = await domain.recordSweepBatch(FEED, [run], work.id); + const batch = await store.readSweepBatch(FEED, batchId); + const group = (await store.readFeed(FEED)).routineActions.find((candidate) => candidate.id === "just-before")!; + await store.writeRoutineActionGroup({ ...group, createdAt: batch.createdAt }); // force the timestamp tie + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: false, routineGroupItems: 0 }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a voice instruction on a card counts as reviewing it", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "voiced" }], { cursor: "voice" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { id: "voiced", title: "Voiced", why: "The user will dictate an instruction.", blocks: [], sourceRunIds: [run] }); + const result = await domain.submitVoiceInstruction(FEED, { kind: "card", feedId: FEED, cardId: "voiced" }, "Reply and ask for the invoice."); + expect("work" in result).toBe(true); + expect((await store.readCard(FEED, "voiced")).status).toBe("queued"); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a card hidden by sweep feedback does not present its judgment", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "hidden-card" }], { cursor: "hidden" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { id: "hidden-card", title: "Hidden", why: "Rejudged away.", blocks: [], sourceRunIds: [run] }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + const card = await store.readCard(FEED, "hidden-card"); + await store.writeCard({ ...card, sweep: { rank: 0, hidden: true, feedbackId: "feedback-1" } }); + const status = await domain.sweepPresentationStatus(FEED); + expect(status.ready).toBe(false); + expect(status.missing[0].reason).toContain("is hidden by sweep feedback feedback-1"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("an older routine group approved after the run is not new coverage", async () => { + const { root, runtime, store, domain } = await setup(); + try { + await domain.upsertRoutineActionGroup(FEED, { + id: "older", label: "Older", summary: "Proposed before this sweep.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "i1", title: "One", reason: "Routine." }], + }); + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "routine_action" }], { cursor: "approved-later" }, work.id); + // Approved (queued) after the run but before the batch, so recording the batch does not stale it. + const approval = await domain.approveRoutineActionGroup(FEED, "older"); + expect((await store.readFeed(FEED)).routineActions.find((group) => group.id === "older")?.status).toBe("queued"); + await domain.recordSweepBatch(FEED, [run], work.id); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: false, routineGroupItems: 0 }); + // Cancelling its work returns the older group to `proposed`; it is still not this sweep's proposal. + await domain.cancelQueuedWork(FEED, approval.id); + expect((await store.readFeed(FEED)).routineActions.find((group) => group.id === "older")?.status).toBe("proposed"); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: false, routineGroupItems: 0 }); + // Explicitly re-proposing it under the same id is a new proposal for this sweep. + await domain.upsertRoutineActionGroup(FEED, { + id: "older", label: "Older", summary: "Re-proposed for this sweep.", + proposedAction: { label: "Archive", instruction: "Archive these." }, + items: [{ id: "i1", title: "One", reason: "Routine." }], + }); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: true, routineGroupItems: 1 }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a re-proposed group that the user approves before completion keeps counting", async () => { + const { root, runtime, store, domain } = await setup(); + try { + await domain.upsertRoutineActionGroup(FEED, { + id: "weekly", label: "Weekly", summary: "Last sweep's proposal.", + proposedAction: { label: "Archive", instruction: "Archive the digests." }, + items: [{ id: "old", title: "Old digest", reason: "Routine." }], + }); + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "routine_action" }], { cursor: "revived" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); // stales the earlier proposal + await domain.upsertRoutineActionGroup(FEED, { + id: "weekly", label: "Weekly", summary: "This sweep's proposal under the same id.", + proposedAction: { label: "Archive", instruction: "Archive the digests." }, + items: [{ id: "new", title: "New digest", reason: "Routine." }], + }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + await domain.approveRoutineActionGroup(FEED, "weekly"); + expect((await store.readFeed(FEED)).routineActions.find((group) => group.id === "weekly")?.status).toBe("queued"); + expect(await domain.sweepPresentationStatus(FEED)).toMatchObject({ ready: true, routineGroupItems: 1 }); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("two judgments may deliberately share one cardId", async () => { + const { root, runtime, store, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }, { b: 2 }], [{ decision: "review", cardId: "merged" }, { decision: "routine_action", cardId: "merged" }], { cursor: "shared" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + await domain.upsertCard(FEED, { id: "merged", title: "Merged", why: "Presents both.", blocks: [], sourceRunIds: [run] }); + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + expect((await domain.completeWork(FEED, work.id, work.capabilityToken, { response: "Done." })).status).toBe("completed"); + expect(await store.readSourceCheckpoint(FEED, SOURCE)).toEqual({ cursor: "shared" }); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("a judgment cardId is validated when the run is recorded", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + await expect(domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "has spaces" }], { cursor: "bad" }, work.id)) + .rejects.toThrow("Judgment 1 cardId must use only letters, numbers, dots, underscores, and hyphens."); + await expect(domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "suppress" }, { decision: "review", cardId: 7 }], { cursor: "bad" }, work.id)) + .rejects.toThrow("Judgment 2 cardId must be a non-empty string."); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + +test("re-claiming an interrupted recollection returns what is still missing", async () => { + const { root, runtime, domain } = await setup(); + try { + const work = await claimRecollection(domain); + const run = await domain.recordSourceRun(FEED, SOURCE, [{ a: 1 }], [{ decision: "review", cardId: "thread-x" }], { cursor: "reclaim" }, work.id); + await domain.recordSweepBatch(FEED, [run], work.id); + // The claiming session dies; a fresh session re-claims the same work through the lane replay path. + const released = await domain.releaseWork(FEED, work.id, work.capabilityToken); + expect(released.status).toBe("queued"); + const reclaimed = await domain.claimWork(FEED, THREAD) as WorkItem; + expect(reclaimed.id).toBe(work.id); + const output = formatWorkClaimOutput(FEED, reclaimed, { sweepPresentation: await domain.sweepPresentationStatus(FEED) }); + if (!("operatorGuidance" in output)) throw new Error("Expected claim guidance."); + expect(output.operatorGuidance?.completionPrerequisite).toContain("present the judgments listed in pendingPresentation.missing, rerun sweep:status"); + expect(output.operatorGuidance?.pendingPresentation?.missing.map((gap) => gap.cardId)).toEqual(["thread-x"]); + expect(output.operatorGuidance?.requiredWriteBack).toContain("cardId"); + // A run recorded after the batch changes the instruction: re-record the batch first. + const late = await domain.recordSourceRun(FEED, SOURCE, [{ later: true }], [{ decision: "suppress" }], { cursor: "late" }, work.id); + const lateOutput = formatWorkClaimOutput(FEED, reclaimed, { sweepPresentation: await domain.sweepPresentationStatus(FEED) }); + if (!("operatorGuidance" in lateOutput)) throw new Error("Expected claim guidance."); + expect(lateOutput.operatorGuidance?.completionPrerequisite).toContain(`${late} recorded afterwards is not in it. Record the batch again with sweep:record-batch --work ${work.id}`); + expect(lateOutput.operatorGuidance?.completionPrerequisite).toContain("then rerun sweep:status"); + await domain.recordSweepBatch(FEED, [late], work.id); // supersedes run and its judgment moves out of the batch + expect((await domain.sweepPresentationStatus(FEED)).ready).toBe(true); + expect((await domain.completeWork(FEED, work.id, reclaimed.capabilityToken, { response: "Finished." })).status).toBe("completed"); + } finally { + runtime.sqlite.close(); + await rm(root, { recursive: true, force: true }); + } +}); + test("runs recorded without recollection work still advance the checkpoint immediately", async () => { const { root, runtime, store, domain } = await setup(); try { diff --git a/test/templates.test.ts b/test/templates.test.ts index 0f645ca..8baa6f2 100644 --- a/test/templates.test.ts +++ b/test/templates.test.ts @@ -1,5 +1,5 @@ import { expect, test } from "bun:test"; -import { COMPOSE_CARD_PROMPT } from "../server/templates"; +import { BASE_JUDGE_PROMPT, COMPOSE_CARD_PROMPT } from "../server/templates"; test("card composition prompt distinguishes local dismissal from source cleanup", () => { expect(COMPOSE_CARD_PROMPT).toContain("`dismiss_card`"); @@ -7,3 +7,9 @@ test("card composition prompt distinguishes local dismissal from source cleanup" expect(COMPOSE_CARD_PROMPT).toContain("explicit source cleanup"); expect(COMPOSE_CARD_PROMPT).toContain("routine “clear this card” control"); }); + +test("the judge prompt tells agents to name each judgment's card and reuse it", () => { + expect(BASE_JUDGE_PROMPT).toContain("stable `cardId`"); + expect(BASE_JUDGE_PROMPT).toContain("reuse that exact id in `card:upsert`"); + expect(BASE_JUDGE_PROMPT).toContain("before the source\ncheckpoint advances"); +});