mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-07 05:56:21 -10:00
fix(daemon): require source evidence to resolve stuck findings (#2153)
An existing stuck-sweep finding now closes only on the source's delivery, pickup or explicit custody evidence. A failed wake becoming unconfirmed, a fresh human-notification post window, an unreadable registry or a larger reporting-age threshold no longer closes a finding just by removing it from a report. Claimed items returned to pending take part in the existing stalled-pickup derivation, and human notifications stay separate from agent wakes. Older human notices without owner-notification history can close on their item-scoped posted receipt when the recipient is known to be human and unchanged: a destination reroute, or a park whose person changed, needs a new matching notification episode, since the old receipt carries no recipient key. Closure notes name the evidence and record stuck-sweep@system as the actor, keeping finding ownership. Refs #1419, following #1449 and #1811; addresses the false findings reported in #2151. 5 files, +515 -19 (source +91 -8, tests +376, docs +48). No new refusals.
This commit is contained in:
1 parent
aaa681c28a
commit
2fbe0ddb49
5 files changed
+515
-19
No files matched your search
@@ -178,15 +178,58 @@ the destination's orchestrator when one is derivable, otherwise to the
|
||||
destination itself, and that seat may be dead. Overdue claims go to their
|
||||
destination. Repeated detections reuse the existing finding without changing
|
||||
its route; when the source condition resolves, the sweep closes its own finding.
|
||||
Previously, falling out of a reporting query could itself close a delivery or
|
||||
unclaimed finding. An unconfirmed retry, a fresh human-notification post window,
|
||||
an unreadable human registry or a larger reporting-age setting could therefore
|
||||
look like successful recovery. These findings now require source evidence:
|
||||
closure, pickup or parking of the work, a verified/accepted wake, or a posted
|
||||
notification for the current human recipient and episode, as appropriate to the
|
||||
finding. `delivered-ack-pending`, `indeterminate:…`, missing results and unrelated
|
||||
legacy human receipts do not prove resolution. Uncertainty keeps an existing
|
||||
finding open without itself sending another wake or creating another finding.
|
||||
A verified wake resolves a wake failure, not an unclaimed obligation: delivery
|
||||
does not prove pickup.
|
||||
|
||||
A legacy human notice with no owner-notification history can still resolve from
|
||||
the existing item-scoped posted ledger when its recipient is confirmed human.
|
||||
This covers older pending asks and parks without inventing an episode that was
|
||||
never recorded. A registered alias needs a readable registry to establish its
|
||||
human recipient; reserved human references remain known during an outage.
|
||||
Legacy receipts carry no recipient key, so a destination reroute requires a new
|
||||
matching notification episode. A legacy park is accepted only when no earlier
|
||||
notice or reroute predates it, and later history contains only receipts or
|
||||
asking-agent reroutes within that park. A re-park, an exit from the park or an
|
||||
ambiguous append cannot establish the same recipient without owner history;
|
||||
the finding waits for a new matching episode or other source resolution.
|
||||
Rerouting only the asking agent within an otherwise unchanged park preserves
|
||||
its human recipient (`blockedOn`).
|
||||
Modern notifications still require their current episode, and an unrelated
|
||||
legacy human receipt cannot resolve an agent wake.
|
||||
|
||||
A known hold still leaves recovery with the existing mechanism: an ask on
|
||||
Steering without an alert, a wake held by the seat typing guard, or a live
|
||||
quiet-release job. Its closure note names that disposition rather than claiming
|
||||
delivery or pickup. An unreadable quiet hold is not positive closure evidence.
|
||||
Claimed work in either `pending` or `in-progress` uses the same pickup derivation;
|
||||
returning a row to `pending` while retaining its claim no longer hides a later
|
||||
stall or closes its existing stalled-claim finding. A retained failed wake cannot
|
||||
override that pickup evidence. The agent's claim still does not establish delivery
|
||||
of a separate notification to a person.
|
||||
|
||||
That automatic closure records `stuck-sweep@system` as the actor in both the
|
||||
transition and the `queue.updated` audit event. Previously it credited the
|
||||
finding's source agent with the detector's action. The finding's source,
|
||||
transition and the `queue.updated` audit event, with the source evidence in the
|
||||
delivery and unclaimed-obligation closure notes. Previously it credited the finding's source agent
|
||||
with the detector's action. The finding's source,
|
||||
destination and claim remain its existing ownership records; the underlying work
|
||||
is not reassigned, and existing transition history is not rewritten.
|
||||
|
||||
`queue-stuck-sweep.test.ts` checks this with migrated SQLite: pending and claimed
|
||||
findings close once, their transition and event name the sweep, and ownership
|
||||
stays intact. The posted-human-notice case also checks the actor while leaving
|
||||
`queue-stuck-sweep-resolution.test.ts` checks unconfirmed retries, retained
|
||||
claims, current human-notification episodes and registry outages with migrated
|
||||
SQLite, including positive recovery controls and no repeated send on an
|
||||
unresolved retry. `quiet-hold-custody.test.ts` checks existing findings with
|
||||
readable and unreadable quiet custody. `queue-stuck-sweep.test.ts` checks that
|
||||
pending and claimed findings close once, name the sweep in their transition and
|
||||
event, and keep their ownership. The posted-human-notice case also checks the actor while leaving
|
||||
the person's original ask open. These checks exercise the detector and queue
|
||||
repository, not a live agent or connector.
|
||||
|
||||
|
||||
@@ -376,6 +376,81 @@ function isFindingRow(item: QueueItem): boolean {
|
||||
return (item.tags ?? []).includes(STUCK_SWEEP_FINDING_TAG);
|
||||
}
|
||||
|
||||
/** Without OWNER history a receipt has no recipient/episode key. A simple
|
||||
* legacy park is still identifiable, but a later state write or an earlier
|
||||
* notice/reroute makes that binding unknown. Even a same-state append may be
|
||||
* a re-park whose recipient could not be classified during a registry outage. */
|
||||
function legacyHumanRecipientRetained(deps: StuckSweepDeps, row: QueueItem): boolean {
|
||||
const history = deps.queueRepo.transitionLog.listForQitem(row.qitemId);
|
||||
const isReceipt = (note: string | null) => note?.startsWith("slack-owner-notification-posted ")
|
||||
|| note?.startsWith("slack-owner-notification-transport-failed ");
|
||||
const isReroute = (note: string | null) => note?.startsWith("fallback-routed:");
|
||||
if (row.state !== "blocked") {
|
||||
return !history.some((transition) => transition.state === "blocked" || isReroute(transition.transitionNote));
|
||||
}
|
||||
const firstPark = history.findIndex((transition) => transition.state === "blocked");
|
||||
if (firstPark < 0 || history.slice(0, firstPark).some((transition) =>
|
||||
isReceipt(transition.transitionNote) || isReroute(transition.transitionNote))) return false;
|
||||
// Receipts and asking-agent reroutes preserve blockedOn. Other appends do
|
||||
// not record whether it changed, so they cannot prove the original park.
|
||||
return history.slice(firstPark + 1).every((transition) => transition.state === "blocked"
|
||||
&& (isReceipt(transition.transitionNote) || isReroute(transition.transitionNote)));
|
||||
}
|
||||
|
||||
/** Falling out of a reporting query is not proof of delivery or pickup. Read the
|
||||
* source's receipts and dispositions before closing an existing finding; an
|
||||
* unavailable source or an unconfirmed send supplies no resolution evidence. */
|
||||
function deliveryResolutionEvidence(
|
||||
deps: StuckSweepDeps,
|
||||
kind: "undelivered-wake" | "undelivered-human-notice" | "unclaimed-obligation",
|
||||
qitemId: string,
|
||||
now: Date,
|
||||
): string | null {
|
||||
const row = deps.queueRepo.getById(qitemId);
|
||||
if (!row) return null;
|
||||
if (["done", "failed", "denied", "canceled", "handed-off"].includes(row.state)) {
|
||||
return `source closed (${row.state})`;
|
||||
}
|
||||
// An agent's claim/park resolves its failed wake or unclaimed work, not the
|
||||
// separate notification owed to a person while that agent is waiting on them.
|
||||
if (kind !== "undelivered-human-notice") {
|
||||
if (row.claimedAt) return `source claimed at ${row.claimedAt}`;
|
||||
if (row.state === "blocked") return "source parked";
|
||||
if (kind === "undelivered-wake") {
|
||||
if (row.lastNudgeResult === "verified") return "source wake verified";
|
||||
if (row.lastNudgeResult?.startsWith("wake-accepted:")) return "source wake accepted by its hosted trigger";
|
||||
}
|
||||
}
|
||||
if (deps.queueRepo.humanNotificationPostedThisEpisode(qitemId)) return "current human ask posted";
|
||||
if (kind === "undelivered-human-notice") {
|
||||
const notification = deps.queueRepo.transitionLog.latestOwnerNotificationForQitem(qitemId);
|
||||
// Park and resolution notices have their own current ledger; a row-scoped
|
||||
// legacy receipt cannot establish delivery of an unrelated agent wake/ask.
|
||||
if (notification && (row.state === "blocked" || notification.ownerNotificationKind === "human-decision-resolved")
|
||||
&& deps.queueRepo.deliveryOutcomeFor(qitemId)?.outcome === "posted") {
|
||||
return "current human notification posted";
|
||||
}
|
||||
// Pre-OWNER notices have no episode to match. Their existing row-scoped
|
||||
// ledger can resolve that human notice, but never an unrelated agent wake.
|
||||
if (!notification && deps.queueRepo.deliveryOutcomeFor(qitemId)?.outcome === "posted") {
|
||||
const recipient = row.state === "blocked" ? row.blockedOn : row.destinationSession;
|
||||
const recipientRetained = legacyHumanRecipientRetained(deps, row);
|
||||
try {
|
||||
if (recipientRetained && recipient && deps.queueRepo.isHumanDestination(recipient)) return "legacy human notification posted";
|
||||
} catch { /* An unreadable registry does not establish the recipient. */ }
|
||||
}
|
||||
}
|
||||
if (row.state === "pending" && deps.queueRepo.humanAskWithoutAlert(qitemId)) return "current human ask placed on Steering without an alert";
|
||||
const quiet = quietHoldCustody(deps.db, qitemId, now);
|
||||
if (quiet && !quiet.unreadable && (quiet.state === "held" || quiet.state === "released")) {
|
||||
return `source held by quiet-release job for window ${quiet.windowId}`;
|
||||
}
|
||||
if (kind !== "undelivered-human-notice" && row.lastNudgeResult === "retained:typing_guard") {
|
||||
return "source wake held by the seat typing guard";
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function latestIso(...values: Array<string | null | undefined>): string {
|
||||
let latest: { iso: string; time: number } | undefined;
|
||||
for (const iso of values) {
|
||||
@@ -456,7 +531,7 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
|
||||
/** The dedup tags of the candidates the LADDER suppressed in this pass, and only those:
|
||||
* a row the ladder alone would have stopped. A finding for such a row stays open,
|
||||
* because the marker only ever stopped the re-report — it never resolved anything.
|
||||
* Every other finding resolves as soon as the sweep no longer detects its condition. */
|
||||
* Other delivery/pickup findings still need source evidence before closure. */
|
||||
const suppressedTags = new Set<string>();
|
||||
const ageMinutes = deps.unclaimedAgeMinutes ?? resolveStuckSweepUnclaimedAgeMinutes();
|
||||
const routeNotes = new Map<string, string>();
|
||||
@@ -497,7 +572,7 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
|
||||
// S04 seam — a claimed row with no later motion past the pickup threshold. The
|
||||
// pickup module remains the ONE derivation rule; this loop only enumerates and routes.
|
||||
const claimedRows = deps.db
|
||||
.prepare("SELECT qitem_id FROM queue_items WHERE state = 'in-progress' AND claimed_at IS NOT NULL")
|
||||
.prepare("SELECT qitem_id FROM queue_items WHERE state IN ('pending', 'in-progress') AND claimed_at IS NOT NULL")
|
||||
.all() as Array<{ qitem_id: string }>;
|
||||
for (const { qitem_id } of claimedRows) {
|
||||
const row = deps.queueRepo.getById(qitem_id);
|
||||
@@ -545,6 +620,10 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
|
||||
for (const row of deps.queueRepo.findUndelivered()) {
|
||||
if (isFindingRow(row)) continue;
|
||||
const humanNotice = row.deliveryOutcome === "never-posted" || row.deliveryOutcome === "transport-failed";
|
||||
// A claim supersedes a retained failed wake, including on a row returned to
|
||||
// pending. Its pickup finding must not be replaced by that historical failure;
|
||||
// a separate human notification still needs its own delivery evidence.
|
||||
if (!humanNotice && row.claimedAt) continue;
|
||||
const notification = humanNotice ? deps.queueRepo.transitionLog.latestOwnerNotificationForQitem(row.qitemId) : null;
|
||||
const resolutionNotice = notification?.ownerNotificationKind === "human-decision-resolved";
|
||||
// Match the notification source in slack/queue-access.ts, not the agent wake's escalator.
|
||||
@@ -880,19 +959,23 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
|
||||
const dedupTag = tags.find((t) => t.startsWith("stuck-sweep:"));
|
||||
if (!dedupTag || liveDedupTags.has(dedupTag)) continue;
|
||||
const [, kind, stuckId] = dedupTag.match(/^stuck-sweep:([a-z-]+):(.+)$/) ?? [];
|
||||
// Suppression is not resolution (#1419), and only the ladder's own suppression counts:
|
||||
// a finding whose row the ladder stopped in THIS pass stays open, because the marker
|
||||
// only ever stopped the re-report. Every other finding resolves as soon as the sweep
|
||||
// no longer detects its condition — a resumed claim, a verified custody disposition, a
|
||||
// notice that has since posted, or a row the ladder never owned.
|
||||
// A retry marker only suppresses reporting. Other reporting filters (a new
|
||||
// unconfirmed send, a fresh post window, or a larger unclaimed age) do not
|
||||
// prove resolution either: delivery/pickup findings need source evidence.
|
||||
if (suppressedTags.has(dedupTag)) continue;
|
||||
let resolution = "no longer detected";
|
||||
if (stuckId && (kind === "undelivered-wake" || kind === "undelivered-human-notice" || kind === "unclaimed-obligation")) {
|
||||
const evidence = deliveryResolutionEvidence(deps, kind, stuckId, now);
|
||||
if (!evidence) continue;
|
||||
resolution = evidence;
|
||||
}
|
||||
await deps.queueRepo.update({
|
||||
qitemId: f.qitem_id,
|
||||
// The sweep performed this transition; ownership stays on the existing row.
|
||||
actorSession: "stuck-sweep@system",
|
||||
state: "done",
|
||||
closureReason: "no-follow-on",
|
||||
transitionNote: `stuck-sweep resolved: ${kind ?? "finding"} on ${stuckId ?? "row"} no longer detected`,
|
||||
transitionNote: `stuck-sweep resolved: ${kind ?? "finding"} on ${stuckId ?? "row"} ${resolution}`,
|
||||
});
|
||||
if (kind && stuckId) {
|
||||
findings.push({
|
||||
|
||||
@@ -0,0 +1,350 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import Database from "better-sqlite3";
|
||||
import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
|
||||
import { migrate } from "../src/db/migrate.js";
|
||||
import { EventBus } from "../src/domain/event-bus.js";
|
||||
import type { HumanFragment } from "../src/domain/gateway/human-registry.js";
|
||||
import { QueueRepository, type QueueItem } from "../src/domain/queue-repository.js";
|
||||
|
||||
describe("stuck-sweep resolution uses source evidence", () => {
|
||||
let db: Database.Database;
|
||||
let repo: QueueRepository;
|
||||
let registryAvailable: boolean;
|
||||
const person: HumanFragment = {
|
||||
entityId: "human-owner", class: "human", displayName: "Owner", address: "human-owner@external",
|
||||
connectorBindings: [{ kind: "slack", connectorRef: "slack:fixture", secretsRef: "vault:fixture", role: "primary" }],
|
||||
prefs: { deliveryClass: "A" },
|
||||
};
|
||||
const otherPerson: HumanFragment = { ...person, entityId: "other-person", displayName: "Other person", address: "other-person@external" };
|
||||
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers({ toFake: ["Date"] });
|
||||
vi.setSystemTime(new Date("2025-01-01T12:00:00.000Z"));
|
||||
vi.stubEnv("OPENRIG_QUEUE_PICKUP_STALL_THRESHOLD_MINUTES", "3");
|
||||
db = new Database(":memory:");
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
registryAvailable = true;
|
||||
repo = new QueueRepository(db, new EventBus(db), {
|
||||
validateRig: () => true,
|
||||
loadHumanRegistry: () => registryAvailable
|
||||
? { ok: true, entities: [person, otherPerson] }
|
||||
: { ok: false, error: "fixture registry unavailable" },
|
||||
});
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
db.close();
|
||||
vi.useRealTimers();
|
||||
vi.unstubAllEnvs();
|
||||
});
|
||||
|
||||
function advance(minutes: number): void {
|
||||
vi.setSystemTime(new Date(Date.now() + minutes * 60_000));
|
||||
}
|
||||
|
||||
const work = () => repo.create({
|
||||
sourceSession: "sender@r", destinationSession: "worker@r", body: "work", nudge: false,
|
||||
});
|
||||
|
||||
async function sweep(unclaimedAgeMinutes = 60) {
|
||||
const { runStuckSweep } = await import("../src/domain/queue-stuck-sweep.js");
|
||||
return runStuckSweep({
|
||||
db, queueRepo: repo, now: new Date(), unclaimedAgeMinutes, ladderSuppressionMinutes: 30,
|
||||
resolveOrchestrator: () => null, sessionConfirmedMissing: () => false,
|
||||
resolveRecoveryAgentSeat: () => null, resolveOperatorSeat: () => null,
|
||||
isRegisteredHost: () => false, log: () => {},
|
||||
});
|
||||
}
|
||||
|
||||
function findings(row: QueueItem): QueueItem[] {
|
||||
return repo.list({ limit: 100 }).filter((item) => item.tags?.includes(`recovery-for:${row.qitemId}`));
|
||||
}
|
||||
|
||||
async function failedWake() {
|
||||
const row = await work();
|
||||
repo.recordNudgeAttempt(row.qitemId, "failed:transport unavailable");
|
||||
expect((await sweep()).outcome).toBe("findings");
|
||||
const [finding] = findings(row);
|
||||
expect(finding).toMatchObject({ state: "pending" });
|
||||
return { row, finding: finding! };
|
||||
}
|
||||
|
||||
it.each(["delivered-ack-pending", "indeterminate:transport timeout", "retained:draft_aware"])(
|
||||
"keeps the existing failed-wake finding when the next attempt is %s, without sending again",
|
||||
async (result) => {
|
||||
const { row, finding } = await failedWake();
|
||||
repo.recordNudgeAttempt(row.qitemId, result);
|
||||
const send = vi.fn(async () => ({ ok: true, verified: false }));
|
||||
repo.attachTransport({ send });
|
||||
const before = repo.transitionLog.listForQitem(finding.qitemId);
|
||||
expect(repo.findUndelivered().some((item) => item.qitemId === row.qitemId)).toBe(false);
|
||||
for (let pass = 0; pass < 2; pass++) {
|
||||
expect((await sweep()).findings).not.toContainEqual(expect.objectContaining({ action: "closed" }));
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
}
|
||||
expect(findings(row)).toHaveLength(1);
|
||||
expect(repo.transitionLog.listForQitem(finding.qitemId)).toEqual(before);
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", claimedAt: null, lastNudgeResult: result });
|
||||
expect(send).not.toHaveBeenCalled();
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps a failed-wake finding when its source no longer has a delivery result", async () => {
|
||||
const { row, finding } = await failedWake();
|
||||
db.prepare("UPDATE queue_items SET last_nudge_result = NULL WHERE qitem_id = ?").run(row.qitemId);
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
});
|
||||
|
||||
it("does not resolve an agent's failed wake from a legacy, unrelated human receipt", async () => {
|
||||
const { row, finding } = await failedWake();
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "daemon@kernel",
|
||||
transitionNote: `slack-owner-notification-posted notification_key=${row.qitemId}:0 level=ALERT message_ts=1.2` });
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted");
|
||||
expect(repo.humanNotificationPostedThisEpisode(row.qitemId)).toBe(false);
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
});
|
||||
|
||||
it.each(["verified", "wake-accepted:webhook fixture answered 202", "claim", "claim-returned-to-pending", "done", "canceled"])("closes a failed-wake finding once the source has %s evidence", async (evidence) => {
|
||||
const { row, finding } = await failedWake();
|
||||
if (evidence === "verified" || evidence.startsWith("wake-accepted:")) repo.recordNudgeAttempt(row.qitemId, evidence);
|
||||
else if (evidence === "claim" || evidence === "claim-returned-to-pending") {
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
|
||||
if (evidence === "claim-returned-to-pending") {
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "pending", transitionNote: "returned to pending" });
|
||||
}
|
||||
}
|
||||
else await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: evidence as "done" | "canceled",
|
||||
...(evidence === "done" ? { closureReason: "no-follow-on" as const } : {}), transitionNote: "work resolved" });
|
||||
expect((await sweep()).findings).toContainEqual(expect.objectContaining({ findingQitemId: finding.qitemId, action: "closed" }));
|
||||
expect(repo.getById(finding.qitemId)).toMatchObject({ state: "done", sourceSession: finding.sourceSession,
|
||||
destinationSession: finding.destinationSession, closureReason: "no-follow-on" });
|
||||
expect(repo.transitionLog.listForQitem(finding.qitemId).at(-1)).toMatchObject({ actorSession: "stuck-sweep@system",
|
||||
transitionNote: expect.stringMatching(/source (wake verified|wake accepted|claimed at|closed)/) });
|
||||
expect((await sweep()).findings).not.toContainEqual(expect.objectContaining({ action: "closed" }));
|
||||
});
|
||||
|
||||
it("leaves an explicitly held wake with its typing guard without sending or claiming it", async () => {
|
||||
const { row, finding } = await failedWake();
|
||||
const send = vi.fn(async () => ({ ok: true, verified: false }));
|
||||
repo.attachTransport({ send });
|
||||
repo.recordNudgeAttempt(row.qitemId, "retained:typing_guard");
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("done");
|
||||
expect(repo.transitionLog.listForQitem(finding.qitemId).at(-1)?.transitionNote).toContain("held by the seat typing guard");
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", claimedAt: null, lastNudgeResult: "retained:typing_guard" });
|
||||
expect(send).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not mistake a larger reporting age for pickup of an already-reported obligation", async () => {
|
||||
const row = await work();
|
||||
advance(61);
|
||||
await sweep();
|
||||
const [finding] = findings(row);
|
||||
expect(finding?.tags).toContain(`stuck-sweep:unclaimed-obligation:${row.qitemId}`);
|
||||
await sweep(120);
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("pending");
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
|
||||
await sweep(120);
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("done");
|
||||
});
|
||||
|
||||
it.each([[false, false], [true, false], [false, true], [true, true]])("detects a stalled claim retained on a pending row (existing finding: %s; retained failed wake: %s)", async (existing, failed) => {
|
||||
const row = await work();
|
||||
if (failed) repo.recordNudgeAttempt(row.qitemId, "failed:transport unavailable");
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
|
||||
advance(5);
|
||||
if (existing) await sweep();
|
||||
const before = findings(row);
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "pending", transitionNote: "returned to pending" });
|
||||
advance(5);
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", claimedAt: expect.any(String), pickup: { state: "stalled-after-claim" } });
|
||||
await sweep();
|
||||
const after = findings(row);
|
||||
expect(after).toHaveLength(1);
|
||||
expect(after[0]).toMatchObject({ state: "pending", tags: expect.arrayContaining([`stuck-sweep:stalled-after-claim:${row.qitemId}`]) });
|
||||
if (existing) expect(after[0]!.qitemId).toBe(before[0]!.qitemId);
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", transitionNote: "continued the work" });
|
||||
await sweep();
|
||||
expect(repo.getById(after[0]!.qitemId)?.state).toBe("done");
|
||||
});
|
||||
|
||||
async function humanNotice() {
|
||||
const row = await repo.create({ sourceSession: "sender@r", destinationSession: "human-owner@external",
|
||||
body: "Choose A or B", summary: "A or B?", evidenceRef: "rig queue show qitem-fixture", humanIntent: "decision", humanAlert: true });
|
||||
const key = `${row.qitemId}:${repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)!.transitionId}`;
|
||||
await receipt(row.qitemId, key, "transport-failed");
|
||||
await sweep();
|
||||
const [finding] = findings(row);
|
||||
expect(finding?.tags).toContain(`stuck-sweep:undelivered-human-notice:${row.qitemId}`);
|
||||
return { row, finding: finding!, key };
|
||||
}
|
||||
|
||||
function receipt(qitemId: string, key: string, outcome: "posted" | "transport-failed") {
|
||||
return repo.update({ qitemId, actorSession: "daemon@kernel",
|
||||
transitionNote: `slack-owner-notification-${outcome} notification_key=${key} level=ALERT ${outcome === "posted" ? "message_ts=1.2" : "detail=fixture-failure"}` });
|
||||
}
|
||||
|
||||
it.each([false, true].flatMap((parked) => [person.address, "other-person@r"].map((recipient) => ({ parked, recipient }))))(
|
||||
"closes a delivered legacy human notice without owner history ($recipient; parked: $parked)", async ({ parked, recipient }) => {
|
||||
const row = parked ? await work() : await repo.create({ sourceSession: "sender@r", destinationSession: recipient,
|
||||
body: "Choose A or B", summary: "A or B?", evidenceRef: "rig queue show qitem-fixture", nudge: false });
|
||||
if (parked) {
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "blocked", blockedOn: recipient,
|
||||
summary: "Choose A or B", evidenceRef: "rig queue show qitem-fixture" });
|
||||
}
|
||||
// Migration 076 adds these columns without backfilling historical notices.
|
||||
db.prepare("UPDATE queue_transitions SET owner_notification_kind = NULL, owner_notification_level = NULL WHERE qitem_id = ?").run(row.qitemId);
|
||||
const key = `${row.qitemId}:0`;
|
||||
await receipt(row.qitemId, key, "transport-failed");
|
||||
await sweep();
|
||||
const [finding] = findings(row);
|
||||
expect(finding?.tags).toContain(`stuck-sweep:undelivered-human-notice:${row.qitemId}`);
|
||||
await receipt(row.qitemId, key, "posted");
|
||||
expect(repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)).toBeNull();
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted");
|
||||
expect(repo.humanNotificationPostedThisEpisode(row.qitemId)).toBe(false);
|
||||
registryAvailable = false;
|
||||
await sweep();
|
||||
// Reserved human references remain known; an alias needs its registry.
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe(recipient === person.address ? "done" : "pending");
|
||||
registryAvailable = true;
|
||||
await sweep();
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("done");
|
||||
expect(repo.transitionLog.listForQitem(finding!.qitemId).at(-1)?.transitionNote).toContain("legacy human notification posted");
|
||||
expect(repo.getById(row.qitemId)?.state).toBe(parked ? "blocked" : "pending");
|
||||
});
|
||||
|
||||
it("keeps a claimed agent's separate human notification until its receipt arrives", async () => {
|
||||
const row = await work();
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "blocked", blockedOn: person.address,
|
||||
summary: "Choose A or B", evidenceRef: "rig queue show qitem-fixture" });
|
||||
const key = `${row.qitemId}:${repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)!.transitionId}`;
|
||||
await receipt(row.qitemId, key, "transport-failed");
|
||||
expect(repo.getById(row.qitemId)?.claimedAt).toEqual(expect.any(String));
|
||||
await sweep();
|
||||
const [finding] = findings(row);
|
||||
expect(finding).toMatchObject({ state: "pending", destinationSession: "worker@r",
|
||||
tags: expect.arrayContaining([`stuck-sweep:undelivered-human-notice:${row.qitemId}`]) });
|
||||
registryAvailable = false;
|
||||
await sweep();
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("pending");
|
||||
registryAvailable = true;
|
||||
await receipt(row.qitemId, key, "posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("done");
|
||||
});
|
||||
|
||||
it("keeps a human-delivery finding during a registry outage, then closes it on the current receipt", async () => {
|
||||
const { row, finding, key } = await humanNotice();
|
||||
registryAvailable = false;
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull();
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
registryAvailable = true;
|
||||
await receipt(row.qitemId, key, "posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("done");
|
||||
expect(repo.getById(row.qitemId)?.state).toBe("pending");
|
||||
});
|
||||
|
||||
it("requires a receipt for the new human episode even before its post window expires", async () => {
|
||||
const { row, finding, key } = await humanNotice();
|
||||
await receipt(row.qitemId, key, "posted");
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "sender@r", state: "blocked", blockedOn: "human-owner@external",
|
||||
summary: "Another decision", evidenceRef: "rig queue show qitem-fixture" });
|
||||
const nextKey = `${row.qitemId}:${repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)!.transitionId}`;
|
||||
expect(nextKey).not.toBe(key);
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull();
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
await receipt(row.qitemId, nextKey, "posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("done");
|
||||
});
|
||||
|
||||
it("does not use the old person's receipt to resolve a rerouted human notice", async () => {
|
||||
const { row, finding, key } = await humanNotice();
|
||||
await receipt(row.qitemId, key, "posted");
|
||||
repo.routeToFallback(row.qitemId, "other-person@external", "ask the other person instead");
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted");
|
||||
expect(repo.humanNotificationPostedThisEpisode(row.qitemId)).toBe(false);
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
});
|
||||
|
||||
it.each([false, true])("does not rebind a legacy receipt after changing the human recipient (late receipt: %s)", async (lateReceipt) => {
|
||||
const { row, finding, key } = await humanNotice();
|
||||
db.prepare("UPDATE queue_transitions SET owner_notification_kind = NULL, owner_notification_level = NULL WHERE qitem_id = ?").run(row.qitemId);
|
||||
if (!lateReceipt) await receipt(row.qitemId, key, "posted");
|
||||
repo.routeToFallback(row.qitemId, otherPerson.address, "ask the other person instead");
|
||||
if (lateReceipt) await receipt(row.qitemId, key, "posted");
|
||||
expect(repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)).toBeNull();
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
// A new episode gives the new recipient its own receipt, unlike the legacy note.
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "sender@r", state: "blocked", blockedOn: otherPerson.address,
|
||||
summary: "The other person's decision", evidenceRef: "rig queue show qitem-fixture" });
|
||||
const nextKey = `${row.qitemId}:${repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)!.transitionId}`;
|
||||
expect(nextKey).not.toBe(key);
|
||||
await receipt(row.qitemId, nextKey, "posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe("done");
|
||||
});
|
||||
|
||||
it.each([false, true].flatMap((parked) => [false, true].map((lateReceipt) => ({ parked, lateReceipt }))))(
|
||||
"does not use a former recipient's legacy receipt after a park during a registry outage (already parked: $parked; late receipt: $lateReceipt)",
|
||||
async ({ parked, lateReceipt }) => {
|
||||
const row = parked ? await work() : (await humanNotice()).row;
|
||||
if (parked) {
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "blocked", blockedOn: person.address,
|
||||
summary: "Choose A or B", evidenceRef: "rig queue show qitem-fixture" });
|
||||
}
|
||||
db.prepare("UPDATE queue_transitions SET owner_notification_kind = NULL, owner_notification_level = NULL WHERE qitem_id = ?").run(row.qitemId);
|
||||
const key = `${row.qitemId}:0`;
|
||||
await receipt(row.qitemId, key, "transport-failed");
|
||||
await sweep();
|
||||
const [finding] = findings(row);
|
||||
expect(finding?.tags).toContain(`stuck-sweep:undelivered-human-notice:${row.qitemId}`);
|
||||
if (!lateReceipt) await receipt(row.qitemId, key, "posted");
|
||||
if (!parked) repo.routeToFallback(row.qitemId, otherPerson.address, "ask the other person instead");
|
||||
registryAvailable = false;
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "blocked", blockedOn: otherPerson.address });
|
||||
expect(repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)).toBeNull();
|
||||
if (lateReceipt) await receipt(row.qitemId, key, "posted");
|
||||
registryAvailable = true;
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("pending");
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "pending" });
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "blocked", blockedOn: otherPerson.address });
|
||||
const current = repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)!;
|
||||
await receipt(row.qitemId, `${row.qitemId}:${current.transitionId}`, "posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("done");
|
||||
},
|
||||
);
|
||||
|
||||
it("keeps the legacy park's human recipient when only the asking agent is rerouted", async () => {
|
||||
const row = await work();
|
||||
repo.claim({ qitemId: row.qitemId, destinationSession: "worker@r" });
|
||||
await repo.update({ qitemId: row.qitemId, actorSession: "worker@r", state: "blocked", blockedOn: person.address,
|
||||
summary: "Choose A or B", evidenceRef: "rig queue show qitem-fixture" });
|
||||
db.prepare("UPDATE queue_transitions SET owner_notification_kind = NULL, owner_notification_level = NULL WHERE qitem_id = ?").run(row.qitemId);
|
||||
const key = `${row.qitemId}:0`;
|
||||
await receipt(row.qitemId, key, "transport-failed");
|
||||
await sweep();
|
||||
const [finding] = findings(row);
|
||||
expect(finding?.tags).toContain(`stuck-sweep:undelivered-human-notice:${row.qitemId}`);
|
||||
repo.routeToFallback(row.qitemId, "other-worker@r", "another agent will follow up");
|
||||
await receipt(row.qitemId, key, "posted");
|
||||
await sweep();
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("done");
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "blocked", blockedOn: person.address, destinationSession: "other-worker@r" });
|
||||
});
|
||||
});
|
||||
@@ -1059,12 +1059,7 @@ describe("S02 standing stuck sweep — both halves, routed findings, quiet-but-o
|
||||
actorSession: "sender@r",
|
||||
transitionNote: `${mod.LADDER_ATTEMPT_PREFIX} 1 failed:tmux session not found`,
|
||||
});
|
||||
await repo.update({
|
||||
qitemId: row.qitemId,
|
||||
actorSession: "daemon@kernel",
|
||||
transitionNote: `slack-owner-notification-posted notification_key=${row.qitemId}:0 level=ALERT message_ts=1.2`,
|
||||
});
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted");
|
||||
repo.recordNudgeAttempt(row.qitemId, "verified");
|
||||
await runSweep();
|
||||
const after = await findingsFor(row.qitemId);
|
||||
expect(after).toHaveLength(1);
|
||||
|
||||
@@ -121,6 +121,31 @@ describe("custody of asks held for quiet hours", () => {
|
||||
expect(ladderTouched(row)).toBe(false);
|
||||
});
|
||||
|
||||
it.each([false, true])("an existing finding needs readable quiet custody before closing (unreadable: %s)", async (unreadable) => {
|
||||
const row = await held();
|
||||
// A finding from before the hold: its continued existence is independent of
|
||||
// whether this pass would report the now-held source as a new failure.
|
||||
const finding = await repo.create({ sourceSession: "asker@r", destinationSession: "orch@r", body: "prior wake failure", nudge: false,
|
||||
tags: ["stuck-sweep-finding", `stuck-sweep:undelivered-wake:${row.qitemId}`, `recovery-for:${row.qitemId}`] });
|
||||
let prepare: ReturnType<typeof vi.spyOn> | undefined;
|
||||
if (unreadable) {
|
||||
const original = db.prepare.bind(db);
|
||||
prepare = vi.spyOn(db, "prepare").mockImplementation(((sql: string) => {
|
||||
if (sql.includes("SELECT 1 FROM watchdog_jobs WHERE policy = 'quiet-release'")) throw new Error("fixture custody read failed");
|
||||
return original(sql);
|
||||
}) as typeof db.prepare);
|
||||
}
|
||||
try {
|
||||
await sweep();
|
||||
expect(repo.getById(finding.qitemId)?.state).toBe(unreadable ? "pending" : "done");
|
||||
if (!unreadable) expect(repo.listTransitions(finding.qitemId).at(-1)?.transitionNote).toContain("held by quiet-release job");
|
||||
expect(posts).toHaveLength(0);
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", claimedAt: null });
|
||||
} finally {
|
||||
prepare?.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("a hold whose release job was stopped from outside (the only way one loses its job) is found, by the sweep and by the ladder", async () => {
|
||||
const row = await held();
|
||||
asBaton(row);
|
||||
|
||||
Reference in new issue
Block a user