mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-07 05:56:21 -10:00
fix(daemon): send wake-ladder retries as OpenRig and pass workflow entry sender (#2229)
Queue wakes from the daemon no longer arrive as "From: <unknown sender>". A wake-ladder retry now names OpenRig (daemon@kernel) as its sender and keeps the row's own source visible in the body ("Row source: <seat>."), and a workflow's first entry wake names the session that started the workflow. Retry timing, caps and admission are unchanged. For #2223.
4 files: source (queue-wake-ladder.ts, workflow-runtime.ts; +7 -2) and tests (+54 -3). No new refusal.
This commit is contained in:
4 files changed
+61
-5
No files matched your search
@@ -1,6 +1,8 @@
|
||||
import { resolveEscalationOwner, ESCALATION_HOP_TAG, hasConsumedEscalationHop } from "./escalation-owner.js";
|
||||
import type { ArbitratedSeatState } from "./activity-taxonomy.js";
|
||||
import { seatRuntimeHomeBySession } from "./seat-runtime-home.js";
|
||||
import { OPENRIG_SENDER } from "../lib/pane-envelope.js";
|
||||
import { renderQueueHandoffNudge } from "./queue-nudge-text.js";
|
||||
import { quietHoldCustody } from "./quiet-hold-custody.js";
|
||||
import { latestHookMovedOn, latestHookPermissionPending, latestHookWaitsOnPerson } from "./agent-activity-store.js";
|
||||
import type { AgentActivity } from "./types.js";
|
||||
@@ -640,7 +642,10 @@ export async function runWakeLadderTick(deps: WakeLadderDeps): Promise<WakeLadde
|
||||
deps.attemptWake ??
|
||||
(async (qitemId: string, target: string): Promise<string> => {
|
||||
// A ladder retry is OpenRig's own scheduling on the row, whoever created it.
|
||||
await deps.queueRepo.maybeNudge(qitemId, target, true, undefined, undefined, "openrig");
|
||||
// It is sent as OpenRig (daemon@kernel); the row's original source stays visible in the body.
|
||||
const row = deps.queueRepo.getById(qitemId);
|
||||
const bodyOverride = `${renderQueueHandoffNudge(qitemId, null)}${row?.sourceSession ? ` Row source: ${row.sourceSession}.` : ""}`;
|
||||
await deps.queueRepo.maybeNudge(qitemId, target, true, OPENRIG_SENDER, bodyOverride, "openrig");
|
||||
return deps.queueRepo.getById(qitemId)?.lastNudgeWireResult ?? "indeterminate:no transport available";
|
||||
});
|
||||
|
||||
|
||||
@@ -866,7 +866,7 @@ export class WorkflowRuntime {
|
||||
);
|
||||
});
|
||||
if (entryQitemId && entryQitemDestinationSession) {
|
||||
await this.queueRepo.maybeNudge(entryQitemId, entryQitemDestinationSession, entryQitemNudge);
|
||||
await this.queueRepo.maybeNudge(entryQitemId, entryQitemDestinationSession, entryQitemNudge, input.createdBySession);
|
||||
}
|
||||
|
||||
const finalInstance = this.instanceStore.getByIdOrThrow(instanceId!);
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { describe, it, expect, beforeEach, afterEach } from "vitest";
|
||||
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
|
||||
import { readFileSync } from "node:fs";
|
||||
import { join, resolve } from "node:path";
|
||||
import type Database from "better-sqlite3";
|
||||
@@ -29,6 +29,7 @@ describe("0.3.0 conveyor starter workflow specs", () => {
|
||||
let db: Database.Database;
|
||||
let eventBus: EventBus;
|
||||
let runtime: WorkflowRuntime;
|
||||
let queueRepo: QueueRepository;
|
||||
|
||||
beforeEach(() => {
|
||||
db = createDb();
|
||||
@@ -44,7 +45,7 @@ describe("0.3.0 conveyor starter workflow specs", () => {
|
||||
]);
|
||||
eventBus = new EventBus(db);
|
||||
db.prepare(`INSERT INTO rigs (id, name) VALUES ('r-conveyor', 'conveyor')`).run();
|
||||
const queueRepo = new QueueRepository(db, eventBus, { validateRig: () => true });
|
||||
queueRepo = new QueueRepository(db, eventBus, { validateRig: () => true });
|
||||
// P34: the W1 seam is fail-closed (MF2) — a nudge-intended terminal
|
||||
// close needs a SAME-DB intent store to make its wake durable.
|
||||
queueRepo.attachOutbox(new OutboxHandler(db));
|
||||
@@ -103,6 +104,18 @@ describe("0.3.0 conveyor starter workflow specs", () => {
|
||||
}
|
||||
});
|
||||
|
||||
it("workflow entry nudge is sent as the instantiating session", async () => {
|
||||
const maybeNudge = vi.spyOn(queueRepo, "maybeNudge").mockResolvedValue();
|
||||
const created = await runtime.instantiate({
|
||||
specPath: CONVEYOR_SPEC,
|
||||
rootObjective: "packet sender",
|
||||
createdBySession: "intake-lead@conveyor",
|
||||
});
|
||||
expect(maybeNudge).toHaveBeenCalledTimes(1);
|
||||
expect(maybeNudge.mock.calls[0]?.[0]).toBe(created.entryQitemId);
|
||||
expect(maybeNudge.mock.calls[0]?.[3]).toBe("intake-lead@conveyor");
|
||||
});
|
||||
|
||||
it("conveyor can run multiple active instances on the same rig", async () => {
|
||||
const first = await runtime.instantiate({
|
||||
specPath: CONVEYOR_SPEC,
|
||||
|
||||
@@ -12,8 +12,9 @@
|
||||
// The ladder's state IS the transition log (AM-P3-F6): attempts, rungs, and suspension are
|
||||
// derived from markers at every tick, so a daemon restart resumes at the exact position —
|
||||
// forgotten-ladder and reset-count are both REDs below.
|
||||
import { describe, it, expect, beforeEach, afterEach } from "vitest";
|
||||
import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
|
||||
import Database from "better-sqlite3";
|
||||
import { renderQueueHandoffNudge } from "../src/domain/queue-nudge-text.js";
|
||||
import { migrate } from "../src/db/migrate.js";
|
||||
import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
|
||||
import { QueueRepository, type QueueItem } from "../src/domain/queue-repository.js";
|
||||
@@ -872,3 +873,40 @@ describe("S01 wake-or-escalate — retry ladder, named rungs, derived suspension
|
||||
expect(store.resolveOne("queue.wake_swap_grace_seconds" as never)).toMatchObject({ value: 180, source: "default" });
|
||||
});
|
||||
});
|
||||
|
||||
describe("wake ladder retry — envelope sender", () => {
|
||||
it("a default ladder retry is sent as OpenRig and keeps the row's source visible in the body", async () => {
|
||||
const db = new Database(":memory:");
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
const repo = new QueueRepository(db, new EventBus(db), { validateRig: () => true });
|
||||
const src = await repo.create({ sourceSession: "sender@r", destinationSession: "relay@r", body: "obligation" });
|
||||
const { created } = await repo.handoff({ qitemId: src.qitemId, fromSession: "relay@r", toSession: "worker@r", nudge: false });
|
||||
db.prepare("UPDATE queue_items SET last_nudge_attempt = ?, last_nudge_result = ? WHERE qitem_id = ?").run(
|
||||
new Date(Date.now() - 10 * 60_000).toISOString(),
|
||||
"failed:tmux session not found",
|
||||
created.qitemId,
|
||||
);
|
||||
const maybeNudge = vi.spyOn(repo, "maybeNudge").mockResolvedValue();
|
||||
const mod = await ladderMod();
|
||||
await mod.runWakeLadderTick({
|
||||
db,
|
||||
queueRepo: repo,
|
||||
resolveOrchestrator: () => "orch@r",
|
||||
retryIntervalSeconds: 300,
|
||||
retryCap: 3,
|
||||
unconfirmedWindowMinutes: 30,
|
||||
swapGraceSeconds: 180,
|
||||
log: () => {},
|
||||
});
|
||||
// The body is the default handoff nudge text, plus only the row-source suffix.
|
||||
expect(maybeNudge).toHaveBeenCalledWith(
|
||||
created.qitemId,
|
||||
"worker@r",
|
||||
true,
|
||||
"daemon@kernel",
|
||||
`${renderQueueHandoffNudge(created.qitemId, null)} Row source: relay@r.`,
|
||||
"openrig",
|
||||
);
|
||||
db.close();
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user