mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-07 05:56:21 -10:00
fix(queue): treat informational human updates as readable correspondence (#2586)
QueueRepository.isRegisteredHumanUpdate(row) is true only for a pending row with human_intent update whose destination resolves to a currently registered person; it reads false for an agent, an unregistered destination or an unavailable registry. The waiting view treats such a row as informational correspondence with no claim backstop, and the stuck sweep skips it as an unclaimed obligation, except when its quiet-hours hold has been orphaned (no live release job), which is still reported as before. The new resolution reason closes only the sweep's own unclaimed-obligation finding, never while the row is quiet-held or has a delivery outcome. A shared recovery for such an update can be relabelled to an undelivered human notice instead of being closed, so a later delivery failure stays visible. Posted and on-Steering rows keep today's paths. This covers the informational-update part of #514, not the rest of it.
This commit is contained in:
1 parent
86c677837b
commit
430b676727
4 files changed
+429
-18
No files matched your search
@@ -4619,6 +4619,17 @@ export class QueueRepository {
|
||||
} catch { return false; }
|
||||
}
|
||||
|
||||
/** #514 — a pending explicit informational update addressed to a currently registered human
|
||||
* (external or canonical alias, regardless of terminal transport) is correspondence for reading,
|
||||
* not an obligation to claim. Registry unavailable/unregistered and agent destinations read false. */
|
||||
isRegisteredHumanUpdate(row: Pick<QueueItem, "state" | "humanIntent" | "destinationSession">): boolean {
|
||||
if (row.state !== "pending" || row.humanIntent !== "update") return false;
|
||||
try {
|
||||
const registry = this.loadHumanRegistryFn();
|
||||
return registry.ok && resolveRegisteredHumanAddress(row.destinationSession, registry.entities) !== null;
|
||||
} catch { return false; }
|
||||
}
|
||||
|
||||
classifyDestinationOf(destinationSession: string): DestinationClass {
|
||||
return classifyDestination(destinationSession, {
|
||||
entities: (() => {
|
||||
@@ -5130,15 +5141,22 @@ export class QueueRepository {
|
||||
// episode (same predicate as runStuckSweep), so the view must not promise that sweep.
|
||||
const posted = view.state === "pending" && this.humanNotificationPostedThisEpisode(qitemId, listed);
|
||||
const onSteering = !posted && view.state === "pending" && this.humanAskWithoutAlert(qitemId, listed);
|
||||
const row = !posted && !onSteering ? this.db.prepare(`SELECT state, ${this.hasHumanIntentColumn ? "human_intent" : "NULL AS human_intent"}, destination_session, last_nudge_result FROM queue_items WHERE qitem_id = ?`).get(qitemId) as { state: QueueItem["state"]; human_intent: QueueItem["humanIntent"]; destination_session: string; last_nudge_result: string | null } | undefined : undefined;
|
||||
const noticeOutcome = listed ? listed.ledger?.outcome : this.deliveryOutcomeFor(qitemId)?.outcome;
|
||||
const localOnly = !!row && noticeOutcome !== "never-posted" && noticeOutcome !== "transport-failed"
|
||||
const row = view.state === "pending" && !onSteering
|
||||
? this.db.prepare(`SELECT state, ${this.hasHumanIntentColumn ? "human_intent" : "NULL AS human_intent"}, destination_session, last_nudge_result FROM queue_items WHERE qitem_id = ?`).get(qitemId) as { state: QueueItem["state"]; human_intent: QueueItem["humanIntent"]; destination_session: string; last_nudge_result: string | null } | undefined
|
||||
: undefined;
|
||||
const humanUpdate = !!row && this.isRegisteredHumanUpdate({ state: row.state, humanIntent: row.human_intent, destinationSession: row.destination_session });
|
||||
const noticeOutcome = !posted && !onSteering ? (listed ? listed.ledger?.outcome : this.deliveryOutcomeFor(qitemId)?.outcome) : undefined;
|
||||
const localOnly = !!row && !posted && !humanUpdate && noticeOutcome !== "never-posted" && noticeOutcome !== "transport-failed"
|
||||
&& this.isLocalHumanDecisionWithoutGateway({ state: row.state, humanIntent: row.human_intent, destinationSession: row.destination_session, lastNudgeResult: row.last_nudge_result });
|
||||
const awaitingHuman = posted || onSteering || localOnly;
|
||||
if (awaitingHuman) {
|
||||
if (awaitingHuman || humanUpdate) {
|
||||
view.nextBackstop = { owner: view.owner, mechanism: localOnly ? "local attention feed (no registered human gateway)" : onSteering
|
||||
? "none (on Steering for the human, with no Slack alert; awaiting their decision)"
|
||||
: "none (posted to the human; awaiting their decision)", dueAt: null, intervalSeconds: null };
|
||||
: humanUpdate
|
||||
? posted
|
||||
? "none (posted to the human; informational correspondence requires no claim)"
|
||||
: "none (informational correspondence to the human; no claim required)"
|
||||
: "none (posted to the human; awaiting their decision)", dueAt: null, intervalSeconds: null };
|
||||
}
|
||||
// Quiet hours: a held ask is owned by its window's release job until it posts.
|
||||
const quiet = awaitingHuman ? null : quietCustodyBackstop(view.owner, quietHoldCustody(this.db, qitemId));
|
||||
@@ -5148,7 +5166,7 @@ export class QueueRepository {
|
||||
}
|
||||
const recovery = readWakeLadderBackstop(this.db, qitemId, target => this.acceptsDestination(target));
|
||||
if (recovery && !localOnly) {
|
||||
if (!awaitingHuman) view.laterBackstop = { ...view.nextBackstop, note: "Conditional safety net; current delivery/recovery ownership is evaluated first." };
|
||||
if (!awaitingHuman && !humanUpdate) view.laterBackstop = { ...view.nextBackstop, note: "Conditional safety net; current delivery/recovery ownership is evaluated first." };
|
||||
view.nextBackstop = recovery;
|
||||
} else if (!awaitingHuman && view.state === "pending" && (this.db.prepare("SELECT last_nudge_result FROM queue_items WHERE qitem_id = ?")
|
||||
.get(qitemId) as { last_nudge_result: string | null } | undefined)?.last_nudge_result === "retained:typing_guard") {
|
||||
|
||||
@@ -449,6 +449,9 @@ function deliveryResolutionEvidence(
|
||||
if (quiet && !quiet.unreadable && (quiet.state === "held" || quiet.state === "released")) {
|
||||
return `source held by quiet-release job for window ${quiet.windowId}`;
|
||||
}
|
||||
if (kind === "unclaimed-obligation" && !quiet && deps.queueRepo.isRegisteredHumanUpdate(row) && !deps.queueRepo.deliveryOutcomeFor(qitemId)) {
|
||||
return "informational update to a registered human requires no claim";
|
||||
}
|
||||
if (kind !== "undelivered-human-notice" && row.lastNudgeResult === "retained:typing_guard") {
|
||||
return "source wake held by the seat typing guard";
|
||||
}
|
||||
@@ -727,6 +730,9 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
|
||||
// hold with no live release job is stuck: nothing will ever release it.
|
||||
const quiet = quietHoldCustody(deps.db, row.qitemId, now);
|
||||
if (quiet && quiet.state !== "orphaned") continue;
|
||||
// An explicit informational update to a currently registered human is for reading, not an obligation to claim,
|
||||
// unless its quiet-hours hold has been orphaned before release.
|
||||
if (!quiet && deps.queueRepo.isRegisteredHumanUpdate(row)) continue;
|
||||
// The seat's typing guard or interruption level held its wake on purpose: it's held and readable at the seat, not
|
||||
// stuck. (A handed-off baton's one held notice is the wake ladder's.)
|
||||
if (row.lastNudgeResult === "retained:typing_guard") continue;
|
||||
@@ -951,7 +957,10 @@ export async function runStuckSweep(deps: StuckSweepDeps): Promise<StuckSweepRes
|
||||
if (sharedRow && isFindingRow(sharedRow) && sharedRow.summary?.startsWith("Stuck sweep: ")) {
|
||||
const currentKind = sharedRow.summary.match(/^Stuck sweep: ([a-z-]+) on /)?.[1];
|
||||
const pickupOrDeadline = (kind: string) => kind === "stalled-after-claim" || kind === "overdue-claim";
|
||||
if (currentKind && currentKind !== c.kind && pickupOrDeadline(currentKind) && pickupOrDeadline(c.kind)) {
|
||||
const humanUpdateNotice = currentKind === "unclaimed-obligation"
|
||||
&& c.kind === "undelivered-human-notice"
|
||||
&& deps.queueRepo.isRegisteredHumanUpdate(c.row);
|
||||
if (currentKind && currentKind !== c.kind && ((pickupOrDeadline(currentKind) && pickupOrDeadline(c.kind)) || humanUpdateNotice)) {
|
||||
const marker = "\n\n--- Original finding (historical) ---\n";
|
||||
const original = sharedRow.body.includes(marker) ? sharedRow.body.split(marker, 2)[1]! : sharedRow.body;
|
||||
const body = evidenceBody(deps.db, c) + marker + original;
|
||||
|
||||
@@ -2077,15 +2077,24 @@ describe("#514 unclaimed sweep and a pending ask posted to a registered human",
|
||||
const BETA = human("human-beta");
|
||||
// A registered human whose id has no human- prefix: its canonical alias (owner@kernel) cannot carry an intent.
|
||||
const PLAIN = human("owner");
|
||||
// Feed-only correspondence owes no Slack post; a missing delivery must not obscure the unclaimed net.
|
||||
const READER: HumanFragment = { entityId: "human-reader", class: "human", displayName: "Reader",
|
||||
address: "human-reader@external", connectorBindings: [], prefs: {} };
|
||||
let db: Database.Database;
|
||||
let repo: QueueRepository;
|
||||
let registeredHumans: HumanFragment[];
|
||||
let registryAvailable: boolean;
|
||||
|
||||
beforeEach(() => {
|
||||
db = new Database(":memory:");
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
registeredHumans = [OWNER, ALPHA, BETA, PLAIN, READER];
|
||||
registryAvailable = true;
|
||||
repo = new QueueRepository(db, new EventBus(db), {
|
||||
validateRig: () => true,
|
||||
loadHumanRegistry: () => ({ ok: true as const, entities: [OWNER, ALPHA, BETA, PLAIN] }),
|
||||
loadHumanRegistry: () => registryAvailable
|
||||
? { ok: true as const, entities: registeredHumans }
|
||||
: { ok: false as const, error: "fixture registry outage" },
|
||||
transport: { send: async () => ({ ok: true, verified: true }) },
|
||||
});
|
||||
});
|
||||
@@ -2244,6 +2253,343 @@ describe("#514 unclaimed sweep and a pending ask posted to a registered human",
|
||||
const unclaimedFinding = async (qitemId: string) => (await findingsFor(qitemId))
|
||||
.filter((f) => (f.tags ?? []).includes(`stuck-sweep:unclaimed-obligation:${qitemId}`));
|
||||
|
||||
function bindHumanPane(destination: string): void {
|
||||
// Isolated topology fixture, as in seedLeadSeat: no native session or pane is contacted.
|
||||
const member = destination.split("@")[0]!;
|
||||
db.prepare("INSERT INTO rigs (id, name) VALUES ('rig-reader', 'kernel')").run();
|
||||
db.prepare("INSERT INTO nodes (id, rig_id, logical_id, runtime, cwd) VALUES ('node-reader', 'rig-reader', ?, 'terminal', '/tmp')").run(member);
|
||||
db.prepare("INSERT INTO sessions (id, node_id, session_name, status, created_at) VALUES ('sess-reader', 'node-reader', ?, 'running', '2026-11-05 00:00:00')").run(destination);
|
||||
db.prepare("INSERT INTO bindings (id, node_id, attachment_type, tmux_session, tmux_pane) VALUES ('bind-reader', 'node-reader', 'tmux', ?, '%1')").run(destination);
|
||||
// Distinguish explicit fixture binding from the empty-topology legacy fallback.
|
||||
expect(repo.classifyDestinationOf("unbound-fixture@kernel").class).toBe("unroutable");
|
||||
expect(repo.classifyDestinationOf(destination).class).toBe("pane-bound");
|
||||
}
|
||||
|
||||
function bindAskingAgentPane(): void {
|
||||
db.prepare("INSERT INTO rigs (id, name) VALUES ('rig-asker', 'r')").run();
|
||||
db.prepare("INSERT INTO nodes (id, rig_id, logical_id, runtime, cwd) VALUES ('node-asker', 'rig-asker', 'asker', 'codex', '/tmp')").run();
|
||||
db.prepare("INSERT INTO sessions (id, node_id, session_name, status) VALUES ('sess-asker', 'node-asker', 'asker@r', 'running')").run();
|
||||
db.prepare("INSERT INTO bindings (id, node_id, attachment_type, tmux_session, tmux_pane) VALUES ('bind-asker', 'node-asker', 'tmux', 'asker@r', '%2')").run();
|
||||
expect(repo.classifyDestinationOf("asker@r").class).toBe("pane-bound");
|
||||
}
|
||||
|
||||
const recoveriesFor = (id: string) => repo.list({ limit: 500 })
|
||||
.filter((item) => item.tags?.includes(`recovery-for:${id}`));
|
||||
|
||||
it.each([
|
||||
{ destination: "human-reader@external", transport: "gateway-routable" },
|
||||
{ destination: "human-reader@kernel", transport: "pane-bound" },
|
||||
])("keeps an aged informational update readable at $destination without creating recovery work", async ({ destination, transport }) => {
|
||||
if (transport === "pane-bound") bindHumanPane(destination);
|
||||
expect(repo.classifyDestinationOf(READER.address)).toMatchObject({ class: "gateway-routable", resolvedHuman: READER.address });
|
||||
expect(repo.classifyDestinationOf(destination).class).toBe(transport);
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination,
|
||||
body: "The requested build finished. No action or decision is needed.", summary: "Build finished",
|
||||
evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false });
|
||||
age(row.qitemId);
|
||||
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(before).toBeDefined();
|
||||
// This person uses the Feed, so there is no missing Slack delivery to recover.
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull();
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before);
|
||||
expect(repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId))
|
||||
.toMatchObject({ body: row.body, state: "pending", humanIntent: "update" });
|
||||
expect(await findingsFor(row.qitemId)).toEqual([]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ destination: "human-reader@external", interruption: "removed" },
|
||||
{ destination: "human-reader@kernel", interruption: "removed" },
|
||||
{ destination: "human-reader@external", interruption: "outage" },
|
||||
{ destination: "human-reader@kernel", interruption: "outage" },
|
||||
])("PR5-F1 closes only obsolete claim work after $interruption registration recovers at $destination", async ({ destination, interruption }) => {
|
||||
if (destination.endsWith("@kernel")) bindHumanPane(destination);
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination,
|
||||
body: "Build finished; no action needed.", summary: "Build status", evidenceRef: "https://example.test/build",
|
||||
humanIntent: "update", nudge: false });
|
||||
age(row.qitemId);
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(stored).toBeDefined();
|
||||
if (interruption === "removed") registeredHumans = registeredHumans.filter((person) => person.entityId !== READER.entityId);
|
||||
else registryAvailable = false;
|
||||
const initial = await sweep();
|
||||
expect(initial.outcome, initial.error).not.toBe("failed");
|
||||
const [finding] = await unclaimedFinding(row.qitemId);
|
||||
expect(finding).toMatchObject({ state: "pending" });
|
||||
// Reading an unresolved or subsequently registered source cannot close recovery work.
|
||||
repo.waitingView(row.qitemId);
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("pending");
|
||||
registeredHumans = [OWNER, ALPHA, BETA, PLAIN, READER];
|
||||
registryAvailable = true;
|
||||
const beforeRead = db.serialize();
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", humanIntent: "update" });
|
||||
expect(repo.list({ destinationSession: destination }).map((item) => item.qitemId)).toContain(row.qitemId);
|
||||
repo.waitingView(row.qitemId);
|
||||
expect(db.serialize()).toEqual(beforeRead);
|
||||
expect(repo.getById(finding!.qitemId)?.state).toBe("pending");
|
||||
const second = await sweep();
|
||||
expect(second.outcome, second.error).not.toBe("failed");
|
||||
expect.soft(second.findings, "F1: sweep closes obsolete unclaimed finding").toContainEqual(
|
||||
expect.objectContaining({ kind: "unclaimed-obligation", findingQitemId: finding!.qitemId, action: "closed" }));
|
||||
expect.soft(repo.getById(finding!.qitemId), "F1: recovery is terminal without claiming correspondence")
|
||||
.toMatchObject({ state: "done", closureReason: "no-follow-on" });
|
||||
const resolution = repo.listTransitions(finding!.qitemId).at(-1)?.transitionNote ?? "";
|
||||
expect.soft(resolution, "F1: resolution names informational correspondence").toMatch(/informational|update/i);
|
||||
expect.soft(resolution).not.toMatch(/source claimed|human (ask|notification) posted|wake verified/i);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
expect(repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId))
|
||||
.toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null });
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull();
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ destination: "human-owner@external", interruption: "removed", outcome: "never-posted" },
|
||||
{ destination: "human-owner@kernel", interruption: "removed", outcome: "never-posted" },
|
||||
{ destination: "human-owner@external", interruption: "outage", outcome: "transport-failed" },
|
||||
{ destination: "human-owner@kernel", interruption: "outage", outcome: "transport-failed" },
|
||||
] as const)("PR2586 represents $outcome delivery after $interruption registration returns at $destination", async ({ destination, interruption, outcome }) => {
|
||||
bindAskingAgentPane();
|
||||
if (destination.endsWith("@kernel")) bindHumanPane(destination);
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination,
|
||||
body: "Build finished; no decision needed.", summary: "Build status", evidenceRef: "https://example.test/build",
|
||||
humanIntent: "update", nudge: false });
|
||||
const notification = repo.transitionLog.latestOwnerNotificationForQitem(row.qitemId)!;
|
||||
expect(notification).toMatchObject({ ownerNotificationKind: "human-update", ownerNotificationLevel: "NOTICE" });
|
||||
const key = currentKey(row.qitemId);
|
||||
if (outcome === "transport-failed") await receipt(row.qitemId, "transport-failed", key);
|
||||
// An unrelated receipt is not evidence that this current episode reached its person.
|
||||
const staleKey = `${row.qitemId}:0`;
|
||||
expect(staleKey).not.toBe(key);
|
||||
await receipt(row.qitemId, "posted", staleKey);
|
||||
age(row.qitemId);
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome);
|
||||
if (interruption === "removed") registeredHumans = registeredHumans.filter((person) => person.entityId !== OWNER.entityId);
|
||||
else registryAvailable = false;
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull();
|
||||
const first = await sweep();
|
||||
expect(first.outcome, first.error).not.toBe("failed");
|
||||
const initial = recoveriesFor(row.qitemId);
|
||||
expect(initial).toHaveLength(1);
|
||||
const prior = initial[0]!;
|
||||
expect(prior).toMatchObject({ state: "pending", tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) });
|
||||
const originalHistory = repo.listTransitions(prior.qitemId);
|
||||
registeredHumans = [OWNER, ALPHA, BETA, PLAIN, READER];
|
||||
registryAvailable = true;
|
||||
expect(currentKey(row.qitemId)).toBe(key);
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome);
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(stored).toBeDefined();
|
||||
let current: QueueItem | undefined;
|
||||
for (const pass of [1, 2]) {
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
const open = recoveriesFor(row.qitemId).filter((item) => ["pending", "in-progress", "blocked"].includes(item.state));
|
||||
expect(open, "one live recovery still owns the unresolved delivery").toHaveLength(1);
|
||||
const next = open[0]!;
|
||||
if (pass === 1) expect(next).toMatchObject({ qitemId: prior.qitemId,
|
||||
destinationSession: prior.destinationSession, state: prior.state });
|
||||
if (current) expect(next).toMatchObject({ qitemId: current.qitemId, state: current.state,
|
||||
destinationSession: current.destinationSession, body: current.body, summary: current.summary });
|
||||
current = next;
|
||||
expect.soft(current.summary, `delivery pass ${pass}: current condition is visible`).toContain("undelivered-human-notice");
|
||||
expect.soft(current.body, `delivery pass ${pass}: current recipient and failure are visible`)
|
||||
.toContain(`human notice ${outcome} to ${destination}`);
|
||||
expect(repo.getById(row.qitemId)!.waiting!.nextBackstop).toMatchObject({ owner: current.destinationSession,
|
||||
mechanism: "queue-recovery:delegated", recovery: { qitemId: current.qitemId, state: current.state } });
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
expect(repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId))
|
||||
.toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null });
|
||||
const history = repo.getById(prior.qitemId)!;
|
||||
if (history.body !== prior.body) expect(history.body).toContain(prior.body);
|
||||
expect(repo.listTransitions(prior.qitemId).slice(0, originalHistory.length)).toEqual(originalHistory);
|
||||
}
|
||||
console.info("PR2586_DELIVERY_TRANSITION", { destination, interruption, outcome, priorOwner: prior.destinationSession,
|
||||
currentOwner: current!.destinationSession, currentSummary: current!.summary, currentState: current!.state });
|
||||
// Only a real current-episode post ends the notice; registration and stale receipts did not.
|
||||
await receipt(row.qitemId, "posted", key);
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("posted");
|
||||
const posted = await sweep();
|
||||
expect(posted.outcome, posted.error).not.toBe("failed");
|
||||
expect(recoveriesFor(row.qitemId).filter((item) => ["pending", "in-progress", "blocked"].includes(item.state))).toEqual([]);
|
||||
expect(repo.getById(prior.qitemId)).toMatchObject({ state: "done" });
|
||||
expect(repo.listTransitions(prior.qitemId).at(-1)).toMatchObject({ state: "done",
|
||||
transitionNote: expect.stringContaining("current human ask posted") });
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null });
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ origin: "sweep", state: "pending" },
|
||||
{ origin: "sweep", state: "in-progress" },
|
||||
{ origin: "sweep", state: "blocked" },
|
||||
{ origin: "authored", state: "pending" },
|
||||
{ origin: "authored", state: "in-progress" },
|
||||
{ origin: "authored", state: "blocked" },
|
||||
] as const)("PR2586 preserves $origin recovery custody in $state during the delivery transition", async ({ origin, state }) => {
|
||||
bindAskingAgentPane();
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: OWNER.address,
|
||||
body: "Build status", summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false });
|
||||
age(row.qitemId);
|
||||
registryAvailable = false;
|
||||
if (origin === "authored") await repo.create({ sourceSession: "asker@r", destinationSession: "recovery-owner@r",
|
||||
body: "Keep this investigation and its author evidence.", summary: "Investigate delivery", nudge: false,
|
||||
tags: [`recovery-for:${row.qitemId}`] });
|
||||
const initial = await sweep();
|
||||
expect(initial.outcome, initial.error).not.toBe("failed");
|
||||
expect(recoveriesFor(row.qitemId)).toHaveLength(1);
|
||||
const prior = recoveriesFor(row.qitemId)[0]!;
|
||||
if (origin === "sweep") expect(prior.tags).toContain(`stuck-sweep:unclaimed-obligation:${row.qitemId}`);
|
||||
else expect(prior.tags).not.toContain("stuck-sweep-finding");
|
||||
if (state !== "pending") repo.claim({ qitemId: prior.qitemId, destinationSession: prior.destinationSession });
|
||||
if (state === "blocked") await repo.update({ qitemId: prior.qitemId, actorSession: prior.destinationSession,
|
||||
state: "blocked", blockedOn: "external:delivery-investigation", transitionNote: "Waiting for the transport investigation" });
|
||||
const owned = repo.getById(prior.qitemId)!;
|
||||
expect(owned.state).toBe(state);
|
||||
const history = repo.listTransitions(prior.qitemId);
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
registryAvailable = true;
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe("never-posted");
|
||||
for (const pass of [1, 2]) {
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
const open = recoveriesFor(row.qitemId).filter((item) => ["pending", "in-progress", "blocked"].includes(item.state));
|
||||
expect(open, `custody pass ${pass}`).toHaveLength(1);
|
||||
expect(open[0]).toMatchObject({ qitemId: owned.qitemId, sourceSession: owned.sourceSession,
|
||||
destinationSession: owned.destinationSession, state: owned.state, claimedAt: owned.claimedAt, blockedOn: owned.blockedOn });
|
||||
if (origin === "authored") expect(open[0]).toMatchObject({ body: owned.body, summary: owned.summary });
|
||||
else if (open[0]!.body !== owned.body) expect(open[0]!.body).toContain(owned.body);
|
||||
expect(repo.listTransitions(prior.qitemId).slice(0, history.length)).toEqual(history);
|
||||
expect(repo.getById(row.qitemId)!.waiting!.nextBackstop).toMatchObject({ owner: owned.destinationSession,
|
||||
recovery: { qitemId: owned.qitemId, state: owned.state } });
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
}
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ destination: "human-reader@external", delivery: "feed-only" },
|
||||
{ destination: "human-reader@kernel", delivery: "feed-only" },
|
||||
{ destination: "human-owner@external", delivery: "posted" },
|
||||
{ destination: "human-owner@kernel", delivery: "posted" },
|
||||
])("PR5-F2 shows $delivery informational correspondence at $destination without claim or decision promises", async ({ destination, delivery }) => {
|
||||
if (destination.endsWith("@kernel")) bindHumanPane(destination);
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: destination,
|
||||
body: "Build finished; no decision requested.", summary: "Build status", evidenceRef: "https://example.test/build",
|
||||
humanIntent: "update", nudge: false });
|
||||
if (delivery === "posted") receipt(row.qitemId, "posted");
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome ?? null).toBe(delivery === "posted" ? "posted" : null);
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(stored).toBeDefined();
|
||||
for (const phase of ["before sweep", "after sweep"]) {
|
||||
if (phase === "after sweep") {
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
}
|
||||
const beforeRead = db.serialize();
|
||||
const shown = repo.getById(row.qitemId)!;
|
||||
const listed = repo.list({ destinationSession: destination }).find((item) => item.qitemId === row.qitemId)!;
|
||||
for (const [surface, view] of [["show", shown.waiting], ["list", listed.waiting], ["waitingView", repo.waitingView(row.qitemId)]] as const) {
|
||||
expect(view, `${phase} ${surface}`).toBeDefined();
|
||||
expect.soft(view!.nextBackstop, `F2: ${phase} ${surface} has no claim deadline`)
|
||||
.toMatchObject({ dueAt: null, intervalSeconds: null });
|
||||
expect.soft(view!.nextBackstop.mechanism, `F2: ${phase} ${surface} is correspondence`)
|
||||
.not.toMatch(/unclaimed|decision/i);
|
||||
expect.soft(view!.laterBackstop?.mechanism ?? "", `F2: ${phase} ${surface} has no later claim promise`).not.toMatch(/unclaimed|decision/i);
|
||||
}
|
||||
expect(db.serialize()).toEqual(beforeRead);
|
||||
expect(shown).toMatchObject({ body: row.body, state: "pending", humanIntent: "update", claimedAt: null });
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
expect(await findingsFor(row.qitemId)).toEqual([]);
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps a registered informational update visibly held by its canonical seat's typing guard", async () => {
|
||||
bindHumanPane("human-reader@kernel");
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: "human-reader@kernel",
|
||||
body: "Build status", summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false });
|
||||
repo.recordNudgeAttempt(row.qitemId, "retained:typing_guard");
|
||||
age(row.qitemId);
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(repo.getById(row.qitemId)!.waiting!.nextBackstop).toMatchObject({ owner: "human-reader@kernel", held: true,
|
||||
mechanism: "held by the seat's typing guard", dueAt: null });
|
||||
await sweep();
|
||||
expect(await findingsFor(row.qitemId)).toEqual([]);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
});
|
||||
|
||||
it.each(["decision", undefined] as const)("keeps an unclaimed human request eligible with intent %s", async (humanIntent) => {
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: READER.address,
|
||||
body: "Choose option A or B.", summary: "A or B?", evidenceRef: "https://example.test/decision",
|
||||
humanIntent, humanAlert: true, nudge: false });
|
||||
age(row.qitemId);
|
||||
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(before).toBeDefined();
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: READER.address,
|
||||
tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) })]);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before);
|
||||
});
|
||||
|
||||
it.each([false, true])("keeps an agent obligation eligible (rerouted informational intent: %s)", async (rerouted) => {
|
||||
const row = rerouted
|
||||
? await repo.create({ sourceSession: "asker@r", destinationSession: READER.address, body: "Build status",
|
||||
summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false })
|
||||
: await repo.create({ sourceSession: "asker@r", destinationSession: "worker@r", body: "Review the build", nudge: false });
|
||||
// Direct creates reject humanIntent on agents. Exercise the supported reroute that retains the original intent.
|
||||
if (rerouted) repo.routeToFallback(row.qitemId, "worker@r", "Agent follow-up required");
|
||||
age(row.qitemId);
|
||||
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(before).toBeDefined();
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ destinationSession: "worker@r", humanIntent: rerouted ? "update" : null });
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: "worker@r",
|
||||
tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) })]);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before);
|
||||
});
|
||||
|
||||
it("does not exempt an update merely because its former human address still looks human", async () => {
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: READER.address, body: "Build status",
|
||||
summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false });
|
||||
registeredHumans = registeredHumans.filter((person) => person.entityId !== READER.entityId);
|
||||
age(row.qitemId);
|
||||
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(before).toBeDefined();
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: READER.address,
|
||||
tags: expect.arrayContaining([`stuck-sweep:unclaimed-obligation:${row.qitemId}`]) })]);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before);
|
||||
});
|
||||
|
||||
it.each(["never-posted", "transport-failed"] as const)("still reports a %s informational human notice to its asking agent", async (outcome) => {
|
||||
const row = await repo.create({ sourceSession: "asker@r", destinationSession: OWNER.address, body: "Build status",
|
||||
summary: "Build status", evidenceRef: "https://example.test/build", humanIntent: "update", nudge: false });
|
||||
if (outcome === "transport-failed") await repo.update({ qitemId: row.qitemId, actorSession: "daemon@kernel",
|
||||
transitionNote: `slack-owner-notification-transport-failed notification_key=${currentKey(row.qitemId)} level=NOTICE detail=fixture-http-500` });
|
||||
// Five minutes exercises delivery recovery without introducing the sixty-minute unclaimed condition.
|
||||
age(row.qitemId, 5);
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome);
|
||||
const before = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(before).toBeDefined();
|
||||
const result = await sweep();
|
||||
expect(result.outcome, result.error).not.toBe("failed");
|
||||
expect(await findingsFor(row.qitemId)).toEqual([expect.objectContaining({ destinationSession: "asker@r",
|
||||
tags: expect.arrayContaining([`stuck-sweep:undelivered-human-notice:${row.qitemId}`]),
|
||||
body: expect.stringContaining(`human notice ${outcome}`) })]);
|
||||
const [recovery] = await findingsFor(row.qitemId);
|
||||
const view = repo.getById(row.qitemId)!.waiting!;
|
||||
expect(view.nextBackstop).toMatchObject({ owner: "asker@r", mechanism: "queue-recovery:delegated",
|
||||
recovery: { qitemId: recovery!.qitemId, state: "pending" } });
|
||||
expect.soft(view.laterBackstop?.mechanism ?? "", "F2: delivery recovery must not retain a later unclaimed promise").not.toContain("unclaimed");
|
||||
await sweep();
|
||||
expect(repo.getById(recovery!.qitemId)?.state).toBe("pending");
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)?.outcome).toBe(outcome);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(before);
|
||||
});
|
||||
|
||||
it("a row whose wake the seat's level or hold held isn't reported as unclaimed: it's held at the seat, not stuck", async () => {
|
||||
const heldRow = await repo.create({ sourceSession: "asker@r", destinationSession: "worker@r", body: "held at a quiet seat", nudge: false });
|
||||
const draftRow = await repo.create({ sourceSession: "asker@r", destinationSession: "worker@r", body: "held by a draft", nudge: false });
|
||||
|
||||
@@ -77,15 +77,17 @@ describe("custody of asks held for quiet hours", () => {
|
||||
return w;
|
||||
}
|
||||
let n = 0;
|
||||
const ask = () => repo.create({ sourceSession: "asker@r", destinationSession: "person@external", body: `Question ${++n}: pick one`,
|
||||
summary: `Pick ${n}`, evidenceRef: `/proof/${n}.md`, humanIntent: "decision", humanAlert: true, nudge: false } as never) as Promise<QueueItem>;
|
||||
const ask = (humanIntent: "decision" | "update" = "decision") => repo.create({ sourceSession: "asker@r", destinationSession: "person@external",
|
||||
body: humanIntent === "decision" ? `Question ${++n}: pick one` : `Build ${++n} finished; no decision needed`,
|
||||
summary: humanIntent === "decision" ? `Pick ${n}` : `Build ${n} finished`, evidenceRef: `/proof/${n}.md`,
|
||||
humanIntent, humanAlert: humanIntent === "decision", nudge: false } as never) as Promise<QueueItem>;
|
||||
const alertFor = async (row: QueueItem) =>
|
||||
(await makeQueuePorts(repo, { loadHumanRegistry: registry } as never).listHumanAlerts({ minimumLevel: "RECORD" })).find((a) => a.qitemId === row.qitemId)!;
|
||||
const quietOn = () => { const now = new Date(); quiet.setUntil("person", { until: new Date(now.getTime() + 24 * 3_600_000), timezone: "UTC", setBy: "t", now }); };
|
||||
const releaseJobs = () => new WatchdogJobsRepository(db).listActive().filter((j) => j.policy === QUIET_RELEASE_POLICY);
|
||||
async function held(): Promise<QueueItem> {
|
||||
async function held(humanIntent: "decision" | "update" = "decision"): Promise<QueueItem> {
|
||||
quietOn();
|
||||
const row = await ask();
|
||||
const row = await ask(humanIntent);
|
||||
const w = wire();
|
||||
const alert = await alertFor(row);
|
||||
w.dispatcher.dispatch(OUTBOUND_OP, alert.destinationSession!, toPayload(alert));
|
||||
@@ -106,12 +108,15 @@ describe("custody of asks held for quiet hours", () => {
|
||||
resolveOrchestrator: () => "orch@r", attemptWake: async () => "verified", log: () => {} });
|
||||
const ladderTouched = (row: QueueItem) => repo.listTransitions(row.qitemId).some((t) => t.actorSession === "wake-ladder@system");
|
||||
|
||||
it("a healthy hold aged past the thresholds is held: no stuck finding, and the waiting view says held by quiet hours until the boundary", async () => {
|
||||
const row = await held();
|
||||
it.each(["decision", "update"] as const)("a healthy %s hold aged past the thresholds is held: no stuck finding, and the waiting view says held by quiet hours until the boundary", async (humanIntent) => {
|
||||
const row = await held(humanIntent);
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
expect(findingFor(await sweep(), row)).toEqual([]);
|
||||
const view = repo.waitingView(row.qitemId)!;
|
||||
expect(view.nextBackstop).toMatchObject({ mechanism: expect.stringMatching(/^quiet-release:/), dueAt: quiet.quietStateFor("person", new Date()).nextBoundary!.at });
|
||||
expect(String((view.nextBackstop as { note?: string }).note)).toMatch(/^Held by quiet hours until /);
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", humanIntent, claimedAt: null });
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
});
|
||||
|
||||
it("a healthy hold isn't counted as picked up, and the wake ladder leaves it to its release job", async () => {
|
||||
@@ -146,17 +151,50 @@ describe("custody of asks held for quiet hours", () => {
|
||||
}
|
||||
});
|
||||
|
||||
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();
|
||||
it.each(["decision", "update"] as const)("a %s 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 (humanIntent) => {
|
||||
const row = await held(humanIntent);
|
||||
asBaton(row);
|
||||
new WatchdogJobsRepository(db).markTerminal(releaseJobs()[0]!.jobId, "test: the job went away");
|
||||
expect(quietHoldCustody(db, row.qitemId)).toMatchObject({ state: "orphaned" });
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
const r = await sweep();
|
||||
expect(findingFor(r, row).length).toBeGreaterThan(0);
|
||||
expect(findingWhy(row)).toMatch(/held for quiet hours .* no release job is live/);
|
||||
expect(r.outcome, r.error).not.toBe("failed");
|
||||
expect.soft(findingFor(r, row).length).toBeGreaterThan(0);
|
||||
expect.soft(findingWhy(row)).toMatch(/held for quiet hours .* no release job is live/);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
expect(posts).toHaveLength(0);
|
||||
await ladder();
|
||||
expect(ladderTouched(row), "an orphaned hold isn't pickup, so the ladder takes it up").toBe(true);
|
||||
});
|
||||
|
||||
it("an orphaned update hold keeps its existing unclaimed finding open without informational resolution", async () => {
|
||||
const row = await held("update");
|
||||
new WatchdogJobsRepository(db).markTerminal(releaseJobs()[0]!.jobId, "test: the job went away");
|
||||
expect(quietHoldCustody(db, row.qitemId)).toMatchObject({ state: "orphaned" });
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull();
|
||||
const finding = await repo.create({ sourceSession: "asker@r", destinationSession: "orch@r", body: "prior unclaimed hold evidence", nudge: false,
|
||||
tags: ["stuck-sweep-finding", `stuck-sweep:unclaimed-obligation:${row.qitemId}`, `recovery-for:${row.qitemId}`] });
|
||||
expect(finding.state).toBe("pending");
|
||||
const history = repo.listTransitions(finding.qitemId);
|
||||
const stored = db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId);
|
||||
// The finding predates a larger reporting window; falling outside it isn't resolution.
|
||||
const now = LATER();
|
||||
expect(Date.parse(row.tsCreated)).toBeGreaterThan(now.getTime() - 180 * 60_000);
|
||||
const r = await runStuckSweep({ db, queueRepo: repo, unclaimedAgeMinutes: 180, now,
|
||||
resolveOrchestrator: () => "orch@r", log: () => {} });
|
||||
expect(r.outcome, r.error).not.toBe("failed");
|
||||
expect(findingFor(r, row).filter((f) => f.action !== "closed")).toEqual([]);
|
||||
expect.soft(repo.getById(finding.qitemId)?.state).toBe("pending");
|
||||
expect.soft(repo.listTransitions(finding.qitemId).slice(history.length)
|
||||
.some((t) => t.transitionNote?.includes("informational update to a registered human requires no claim"))).toBe(false);
|
||||
expect(repo.getById(finding.qitemId)?.body).toContain(finding.body);
|
||||
expect(repo.listTransitions(finding.qitemId).slice(0, history.length)).toEqual(history);
|
||||
expect(db.prepare("SELECT * FROM queue_items WHERE qitem_id = ?").get(row.qitemId)).toEqual(stored);
|
||||
expect(repo.getById(row.qitemId)).toMatchObject({ state: "pending", humanIntent: "update", claimedAt: null });
|
||||
expect(repo.deliveryOutcomeFor(row.qitemId)).toBeNull();
|
||||
expect(posts).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("a release dispatched but never posted whose job is gone is orphaned and found, not left waiting for transport", async () => {
|
||||
const row = await held();
|
||||
quiet.end("person", { endedBy: "t", now: new Date() });
|
||||
|
||||
Reference in new issue
Block a user