mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-12 18:48:31 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8f44641637 | ||
|
|
b1460a11d9 | ||
|
|
d4855cbbec | ||
|
|
b5df49407c | ||
|
|
f46bd7ec08 | ||
|
|
c411603c8c | ||
|
|
39f83a7904 |
No files matched your search
+3
-3
@@ -24,9 +24,9 @@ committed change. Two repairs aren't in this release: see Known limits.
|
||||
|
||||
### Before you upgrade
|
||||
|
||||
- **Back up the database, then restart the daemon:** `rig daemon stop`, copy the
|
||||
database file (`rig config get db.path` prints its path), then
|
||||
`rig daemon start`. Your agents keep running through a daemon restart; don't
|
||||
- **Back up the database, install 0.7.13, then restart the daemon:** `rig daemon stop`,
|
||||
copy the database file (`rig config get db.path` prints its path), install
|
||||
OpenRig 0.7.13 using your existing installation method, then `rig daemon start`. Your agents keep running through a daemon restart; don't
|
||||
use `rig down`, which ends them. Three migrations run on the daemon's first start: 160 and 161 (for the experimental hub connector below)
|
||||
and 162 (indexes for session reads). Going back to 0.7.12 after them isn't
|
||||
established. ([#2993](https://github.com/mvschwarz/openrig/pull/2993), [#2997](https://github.com/mvschwarz/openrig/pull/2997))
|
||||
|
||||
@@ -423,7 +423,7 @@ Each startup file and action specifies when it applies:
|
||||
- a launch replaces it, as launches always did, and names it (`guidance_block_refreshed_unrecorded`);
|
||||
- an exact resume or a handover leaves it, and names it as possibly stale (`guidance_block_may_be_stale`).
|
||||
|
||||
**A file keeps one copy of each distinct block text, and nothing you wrote is removed.**
|
||||
**A file keeps one copy of each distinct block text, and tidying the copies never removes a text you wrote.**
|
||||
- **Touched markers:** a marker an editor or formatter rewrapped, with spaces or line breaks inside the comment, still reads as its block, and the next write puts it back in OpenRig's form.
|
||||
- **Collected copies:** a copy that repeats an earlier one exactly is removed. Beside a copy you edited, OpenRig's own unedited copies are removed too, and every different edited copy stays (`guidance_block_copies_removed`).
|
||||
- **A marker that lost its pair:** OpenRig can't tell where that block starts or ends, so it adds no copy beside it and names it (`guidance_block_markers_damaged`). Put back the missing marker or remove the partial block, and OpenRig's next write treats that block like any other. `rig package install` leaves such a file as it is, names it in its result (`notes`), and installs the rest. A marker without its pair beside a complete block is named too (`guidance_block_marker_unpaired`), and the block is written.
|
||||
|
||||
@@ -143,7 +143,8 @@ export function downCommand(depsOverride?: DownDeps): Command {
|
||||
const cmd = new Command("down").description("Tear down a rig")
|
||||
.addHelpText("after", `
|
||||
rig down stops every agent OpenRig started in the rig: their sessions end, and any work in progress stops
|
||||
with them. A session OpenRig didn't start (adopted, bound or discovered) is left running, and rig down names it.
|
||||
with them. For a community harness only its pane stops: its own processes may keep running, and rig down
|
||||
names it. A session OpenRig didn't start (adopted, bound or discovered) is left running, and rig down names it.
|
||||
Before running it, check who is working (rig ps --nodes --rig <rig>) and that the person asked for it;
|
||||
don't stop a team you didn't start unless asked. OpenRig snapshots the rig first (best-effort), and
|
||||
rig up <rig> --existing brings it back, resuming each seat's conversation where its harness allows. With
|
||||
|
||||
@@ -14,6 +14,12 @@ import { runFrontDoor, resolveTuiPath, probeFrontDoor, type FrontDoorIo } from "
|
||||
import { diagnoseRuntimePosture, observeClaudePermission } from "../../daemon/src/domain/permission-drift.js";
|
||||
import { DaemonClient, DaemonConnectionError, DaemonResponseError, DaemonTimeoutError } from "../src/client.js";
|
||||
|
||||
// The compiled front door's spawns guard against a hang (the piped TUI never exiting). A hang never ends, so a budget well
|
||||
// past a slow start still catches one: loaded CI runners took just over 5 s to start the CLI, and 30 s leaves room.
|
||||
const HANG_LIMIT_MS = 30_000;
|
||||
/** Those tests outlive their spawn's budget, so the spawn's own result is what reports a hang. */
|
||||
const HANG_TEST_TIMEOUT_MS = HANG_LIMIT_MS + 10_000;
|
||||
|
||||
// A GUARANTEED-refused local port (bind an ephemeral port, capture it, close → connecting refuses).
|
||||
// A REAL socket refusal, not a hand-built error — buys the actual Node/undici rejection shape.
|
||||
async function refusedPort(): Promise<number> {
|
||||
@@ -487,7 +493,7 @@ describe("PUBLIC bin ownership (guard finding 1 — the wrapper is the real fron
|
||||
const started = Date.now();
|
||||
const result = spawnSync(process.execPath, [binWrapper], {
|
||||
input: "x\n",
|
||||
timeout: 5000,
|
||||
timeout: HANG_LIMIT_MS,
|
||||
encoding: "utf-8",
|
||||
});
|
||||
expect(result.status, JSON.stringify({
|
||||
@@ -496,13 +502,13 @@ describe("PUBLIC bin ownership (guard finding 1 — the wrapper is the real fron
|
||||
stdout: result.stdout?.slice(-4096), stderr: result.stderr?.slice(-4096),
|
||||
})).toBe(0);
|
||||
expect(`${result.stderr}${result.stdout}`).toMatch(/Usage: rig/);
|
||||
});
|
||||
}, HANG_TEST_TIMEOUT_MS);
|
||||
|
||||
it.skipIf(!existsSync(binWrapper))("PUBLIC bin subcommands are untouched (rig --version via wrapper)", () => {
|
||||
const result = spawnSync(process.execPath, [binWrapper, "--version"], { timeout: 5000, encoding: "utf-8" });
|
||||
const result = spawnSync(process.execPath, [binWrapper, "--version"], { timeout: HANG_LIMIT_MS, encoding: "utf-8" });
|
||||
expect(result.status).toBe(0);
|
||||
expect(result.stdout.trim()).toMatch(/^\d+\.\d+\.\d+/);
|
||||
});
|
||||
}, HANG_TEST_TIMEOUT_MS);
|
||||
});
|
||||
|
||||
describe("script-safety integration (the compiled front door)", () => {
|
||||
@@ -511,22 +517,22 @@ describe("script-safety integration (the compiled front door)", () => {
|
||||
const started = Date.now();
|
||||
const result = spawnSync(process.execPath, [cliEntry], {
|
||||
input: "x\n",
|
||||
timeout: 5000,
|
||||
timeout: HANG_LIMIT_MS,
|
||||
encoding: "utf-8",
|
||||
env: { ...process.env, OPENRIG_URL: "", OPENRIG_PORT: "" },
|
||||
});
|
||||
expect(Date.now() - started).toBeLessThan(5000);
|
||||
expect(Date.now() - started).toBeLessThan(HANG_LIMIT_MS);
|
||||
// pre-slice-17 baseline: commander's clean-help path prints usage and
|
||||
// exits 0 — the front door PRESERVES the piped path byte-for-byte
|
||||
expect(result.status).toBe(0);
|
||||
expect(`${result.stderr}${result.stdout}`).toMatch(/Usage: rig/);
|
||||
expect(`${result.stderr}${result.stdout}`).not.toMatch(/\n\s+at /);
|
||||
});
|
||||
}, HANG_TEST_TIMEOUT_MS);
|
||||
|
||||
it.skipIf(!existsSync(cliEntry))("`rig --help` still exits 0 with the full usage (front-door regression)", () => {
|
||||
const result = spawnSync(process.execPath, [cliEntry, "--help"], { timeout: 5000, encoding: "utf-8" });
|
||||
const result = spawnSync(process.execPath, [cliEntry, "--help"], { timeout: HANG_LIMIT_MS, encoding: "utf-8" });
|
||||
expect(result.status).toBe(0);
|
||||
expect(result.stdout).toMatch(/Usage: rig/);
|
||||
expect(result.stdout).toMatch(/daemon/);
|
||||
});
|
||||
}, HANG_TEST_TIMEOUT_MS);
|
||||
});
|
||||
@@ -14,6 +14,13 @@ interface Result {
|
||||
stderr: string;
|
||||
}
|
||||
|
||||
// A create waits on nothing (each claim is one exclusive file create), so one still running after a minute is hung; a
|
||||
// slow start on a loaded CI runner took over 15 s. A batch's creates run together, so the batch tests allow each batch the
|
||||
// full limit plus one for setup and checks, and a hung create there is reported by its own limit. A single-create test
|
||||
// keeps the suite's default timeout, which reports a hang first.
|
||||
const CREATE_LIMIT_MS = 60_000;
|
||||
const batchesTimeout = (batches: number) => (batches + 1) * CREATE_LIMIT_MS;
|
||||
|
||||
describe("public slice creation ordinal claims", () => {
|
||||
let root: string;
|
||||
let workspace: string;
|
||||
@@ -49,7 +56,7 @@ describe("public slice creation ordinal claims", () => {
|
||||
];
|
||||
try {
|
||||
const result = await promisify(execFile)(process.execPath, args, {
|
||||
env: ownedEnv, encoding: "utf8", timeout: 15000,
|
||||
env: ownedEnv, encoding: "utf8", timeout: CREATE_LIMIT_MS,
|
||||
});
|
||||
return { exitCode: 0, ...result };
|
||||
} catch (error) {
|
||||
@@ -95,7 +102,7 @@ describe("public slice creation ordinal claims", () => {
|
||||
expect(fs.readdirSync(path.join(mission, "slices"), { withFileTypes: true }).filter((entry) => entry.isDirectory())).toHaveLength(33);
|
||||
expect(fs.readFileSync(path.join(mission, "README.md"), "utf8")).toBe(original);
|
||||
expect(fs.existsSync(path.join(mission, ".openrig-slice-create-lock"))).toBe(false);
|
||||
}, 30000);
|
||||
}, batchesTimeout(4));
|
||||
|
||||
it("twenty eight-process manifest-backed batches preserve the admitted concurrency contract", async () => {
|
||||
const ref = "slices/01-existing/slice.yaml";
|
||||
@@ -144,7 +151,7 @@ describe("public slice creation ordinal claims", () => {
|
||||
expect(new Set(successful(results).map((slice) => slice.id)).size).toBe(160);
|
||||
expect(fs.readFileSync(path.join(unrelated, "PROOF.md"), "utf8")).toBe("# unrelated authored proof\n");
|
||||
expect(fs.existsSync(path.join(mission, ".openrig-slice-create-lock"))).toBe(false);
|
||||
}, 180000);
|
||||
}, batchesTimeout(20));
|
||||
|
||||
it("a later create repairs only pending complete registrations and preserves authored membership", async () => {
|
||||
const ref = "slices/01-existing/slice.yaml";
|
||||
|
||||
+4
-4
@@ -163,8 +163,8 @@ is not a restore request.
|
||||
1. **Check for a hold first.** Read the restore request, the per-seat instruction file
|
||||
(`<OPENRIG_HOME>/compaction/post-compact-extra/<session>.md`, named after your full session, such as
|
||||
`dev-impl@my-rig` with `.md` added) when it exists, the newest row or message from whoever routes your work, and any hold
|
||||
from the authority above them. A hold, a release order or an operator's own restore map overrides the default
|
||||
order below. Before any write, also run `rig whoami --json` and `rig queue whoami`.
|
||||
from the authority above them. A hold you verified as live, a release order or an operator's own restore map overrides
|
||||
the default order below for the work it covers; do the unaffected work. Before any write, also run `rig whoami --json` and `rig queue whoami`.
|
||||
2. **Name your class and state its read budget** before reading (see "Two restore classes"), with a
|
||||
checkpoint at each step below. The budget exists so that the restore leaves room for the work it was
|
||||
restored to do. Step 5 takes a high-context seat to its Tier 2 line; the extra sources under "Two restore classes" apply only
|
||||
@@ -218,7 +218,7 @@ The audit message asks for a read-depth table and tells you not to conserve toke
|
||||
- where you are in time: what earlier windows established, what the last window did, what is true now;
|
||||
- the connections you have rebuilt, in a few lines;
|
||||
- the connections you could not rebuild, and where you would look;
|
||||
- the next authorized step and who authorizes it. If a hold stands, the next step is waiting.
|
||||
- the next authorized step and who authorizes it. If a hold you verified as live covers it, wait on that; do the unaffected work.
|
||||
|
||||
The next restore map links this note, which keeps the chain unbroken.
|
||||
|
||||
@@ -238,7 +238,7 @@ unaffected work.
|
||||
- Do not launch a fresh session in place of restoring.
|
||||
- An honest `PARTIAL` with its reason is a correct outcome. Claiming coverage you did not reach is the
|
||||
failure.
|
||||
- Do not resume task work until the read-depth table and the reconnect note exist.
|
||||
- Write the read-depth table and the reconnect note as you resume; don't hold unaffected work for them.
|
||||
|
||||
## Failure modes
|
||||
|
||||
|
||||
@@ -494,7 +494,29 @@ export class TmuxAdapter {
|
||||
if (probe.state !== "present") return false;
|
||||
const panes = await this.listPanes(target.session);
|
||||
return panes.length === 1 && panes[0]!.id === target.pane && await this.isPaneDead(target.pane);
|
||||
});
|
||||
}, target => this.lastClientActivity(target.session));
|
||||
}
|
||||
|
||||
/** When a writable tmux client attached to this session last had activity (epoch ms, whole seconds), or null when no
|
||||
* writable client is attached. tmux's client_activity is session-wide and covers more than text keys, so this is
|
||||
* recent writable-client activity, not proof of typing in the pane. A read-only client (a viewer tile) can't type
|
||||
* into the seat and doesn't count. An attach counts: tmux can't tell a key in the attach's second from the attach,
|
||||
* so a delivery can be held briefly after one. A tmux failure or an unreadable row throws: the guard then holds
|
||||
* the paste rather than take it for "no activity" (#3076). */
|
||||
async lastClientActivity(session: string): Promise<number | null> {
|
||||
const exact = exactTarget(session, "session");
|
||||
const format = `#{client_readonly}${TMUX_FIELD_SEPARATOR}#{client_activity}`;
|
||||
const output = await this.run(["tmux", "list-clients", "-t", exact, "-F", format],
|
||||
`tmux list-clients -t ${shellQuote(exact)} -F "${format}"`);
|
||||
let latest: number | null = null;
|
||||
for (const line of output.split("\n").map(row => row.trim()).filter(row => row !== "")) {
|
||||
const [readonly, activity, ...rest] = line.split(TMUX_FIELD_SEPARATOR);
|
||||
if (rest.length || !/^[01]$/.test(readonly ?? "") || !/^[1-9]\d*$/.test(activity ?? "")) {
|
||||
throw new Error(`tmux list-clients: unreadable client row for ${session}`);
|
||||
}
|
||||
if (readonly === "0") latest = Math.max(latest ?? 0, Number(activity) * 1000);
|
||||
}
|
||||
return latest;
|
||||
}
|
||||
private readonly freshProbes = new Map<string, string>();
|
||||
private readonly freshManaged = new Map<string, {nodeId: string; pane: string}>();
|
||||
@@ -1081,12 +1103,14 @@ export class TmuxAdapter {
|
||||
// A paste the guard proves it left unsubmitted (#2429) is submitted first, under this lease, once this send's own
|
||||
// checks pass; they then aren't run a second time.
|
||||
beforePaste: async () => {
|
||||
let checked = false;
|
||||
const checks = async () => { await options?.beforePaste?.(); beforeWrite(); beforeInput?.(); checked = true; };
|
||||
// This send's own checks can wait (a conversation lookup, an input capture), so they run first: the guard's
|
||||
// read of the input is then the last wait before the paste (#3076). A check that refuses leaves a leftover's
|
||||
// record for the next delivery.
|
||||
await options?.beforePaste?.();
|
||||
const checks = async () => { beforeWrite(); beforeInput?.(); };
|
||||
// The synchronous checks run again just before the key: the guard's re-read awaits after `checks`.
|
||||
const enter = async () => { beforeWrite(); beforeInput?.(); return (await this.sendKeysUnchecked(pane, ["Enter"])).ok; };
|
||||
if (checkDraft) await guard!.beforeInput(identity, false, text, { checks, enter });
|
||||
if (!checked) await options?.beforePaste?.();
|
||||
},
|
||||
afterPaste: async () => { if (checkDraft) await guard!.notePasted(identity, text); },
|
||||
} : undefined;
|
||||
|
||||
@@ -105,6 +105,7 @@ import { armQueueWait, backOffQueueWait, refreshQueueWaits, evaluateQueueWait, r
|
||||
import { parseHumanQuestions, unansweredQuestions, describeTypedReplyPlacement, formatHumanAnswers, humanAnswerText, type HumanQuestion, type HumanAnswers, type TypedHumanReply, type RecordHumanAnswerResult } from "./human-questions.js";
|
||||
import { slackMicros } from "./gateway/slack/state-store.js";
|
||||
import { askBackRoute, RUNNER_CLOSE_WAITS, terminalAnswerMeaning } from "./receiver-context.js";
|
||||
import { runSyncSite, runSyncSiteWhenSlow } from "./sync-site-wrap.js";
|
||||
|
||||
/** A direct ask's answer row ID: from the ask, or from the ask and the park episode its answer belongs to. */
|
||||
function humanAnswerId(askId: string, episodeKey?: string): string {
|
||||
@@ -821,14 +822,14 @@ export function createLinkedSuccessorLookup(db: Database.Database, sourceQitemId
|
||||
const startedAt = performance.now();
|
||||
const map = new Map<string, Array<{ qitemId: string; destination: string | null }>>();
|
||||
const ids = JSON.stringify([...sources]);
|
||||
const rows = db.prepare(
|
||||
const rows = runSyncSite("queue.linked_successors.build", () => db.prepare(
|
||||
`SELECT s.qitem_id, s.destination_session, s.handed_off_from AS source FROM queue_items s
|
||||
WHERE s.handed_off_from IN (SELECT value FROM json_each(?))
|
||||
UNION
|
||||
SELECT s.qitem_id, s.destination_session, j.value AS source FROM queue_items s, json_each(s.chain_of_record) j
|
||||
WHERE s.chain_of_record IS NOT NULL AND json_valid(s.chain_of_record)
|
||||
AND j.value IN (SELECT value FROM json_each(?))`,
|
||||
).all(ids, ids) as Array<{ qitem_id: string; destination_session: string | null; source: string }>;
|
||||
).all(ids, ids)) as Array<{ qitem_id: string; destination_session: string | null; source: string }>;
|
||||
for (const r of rows) {
|
||||
const list = map.get(r.source);
|
||||
const entry = { qitemId: r.qitem_id, destination: r.destination_session };
|
||||
@@ -2634,8 +2635,13 @@ export class QueueRepository {
|
||||
}
|
||||
|
||||
/** {@link handoff}'s whole act, synchronously: its transaction commits before this returns, and the wake is delivered
|
||||
* after commit without being awaited, so a synchronous caller (an answer's close path) can hand a row on. */
|
||||
* after commit without being awaited, so a synchronous caller (an answer's close path) can hand a row on. Named in
|
||||
* the slow-operation log, so a stall during a handoff is attributed to it. */
|
||||
private handoffNow(input: QueueHandoffInput): { closed: QueueItem; created: QueueItem } {
|
||||
return runSyncSite("queue.handoff", () => this.handoffWithin(input));
|
||||
}
|
||||
|
||||
private handoffWithin(input: QueueHandoffInput): { closed: QueueItem; created: QueueItem } {
|
||||
const source = this.getById(input.qitemId);
|
||||
if (!source) {
|
||||
throw new QueueRepositoryError(
|
||||
@@ -4563,8 +4569,9 @@ export class QueueRepository {
|
||||
typeof opts?.excludeOperationReply === "string" && opts.excludeOperationReply.length > 0
|
||||
? mcOperationTag(opts.excludeOperationReply)
|
||||
: null;
|
||||
// Named only when slow: it runs on every row read and on each row of the attention read.
|
||||
const replies = this.hasReplyToColumn
|
||||
? (excludeOp
|
||||
? runSyncSiteWhenSlow("queue.decision_revision.replies", () => excludeOp
|
||||
? (this.db
|
||||
.prepare(
|
||||
"SELECT qitem_id FROM queue_items WHERE reply_to = ? AND NOT (tags IS NOT NULL AND json_valid(tags) AND EXISTS (SELECT 1 FROM json_each(tags) WHERE value = ?)) ORDER BY ts_created, rowid",
|
||||
|
||||
@@ -60,6 +60,15 @@ export interface TypingGuardConfig extends TypingGuardSettings { revision: strin
|
||||
const DEFAULT_GUARD: TypingGuardConfig = { mode: "off", revision: "legacy-off" };
|
||||
/** How long the guard waits before reading again an input it couldn't read right after its own paste or Enter (#2429). */
|
||||
const OWN_INPUT_REREAD_MS = 300;
|
||||
/** Right before an automatic paste, the empty input is read once more after this wait: a draft typed (or redrawn by the
|
||||
* agent's TUI) since the first read is seen before the paste, not after it, where nothing is removed (#2447, #3076). */
|
||||
const PASTE_SETTLE_MS = 300;
|
||||
/** Recent activity at a writable tmux client of the seat (keys, but also other client input; session-wide, not proof of
|
||||
* typing in this pane) holds the paste: a later attempt can deliver once it reads quiet and the input empty. tmux
|
||||
* reports client activity in whole seconds, so this is three of them (#3076). */
|
||||
const INPUT_QUIET_MS = 3000;
|
||||
/** tmux's client_activity is in whole seconds; the activity may have come up to this much after the second it reports. */
|
||||
const CLIENT_ACTIVITY_PRECISION_MS = 999;
|
||||
/** A terminal member with nothing set holds OpenRig's own notices and takes deliberate sends. */
|
||||
const TERMINAL_DEFAULT: TypingGuardConfig = { mode: "off", level: "sends-only", revision: "terminal-default" };
|
||||
/** The actor a guard change records when its request carries no seat identity, as from an ordinary terminal (#2350).
|
||||
@@ -122,6 +131,9 @@ export class SeatDeliveryGuard {
|
||||
/** A paste this guard kept from being submitted because it couldn't read the input, per node (#2429). */
|
||||
private readonly leftovers = new Map<string, { target: GuardTarget; humanEpoch: number; text: string; collapsedPaste: string | null }>();
|
||||
private inspectInput?: (target: GuardTarget, owned?: ComposerOwnership) => Promise<ComposerInput | null>;
|
||||
/** When a writable tmux client of the seat last had activity (epoch ms, whole seconds); null when no writable client
|
||||
* is attached; a rejection when that can't be known. */
|
||||
private inputActivity?: (target: GuardTarget) => Promise<number | null>;
|
||||
private inputAbsent?: (target: GuardTarget) => Promise<boolean>;
|
||||
private readonly hasModeConfig: boolean;
|
||||
|
||||
@@ -133,9 +145,11 @@ export class SeatDeliveryGuard {
|
||||
this.hasModeConfig = (db.prepare("PRAGMA table_info(seat_delivery_guards)").all() as Array<{ name: string }>).some(c => c.name === "desired_config");
|
||||
}
|
||||
|
||||
attachInputInspection(inspect: (target: GuardTarget, owned?: ComposerOwnership) => Promise<ComposerInput | null>, absent: (target: GuardTarget) => Promise<boolean>): void {
|
||||
attachInputInspection(inspect: (target: GuardTarget, owned?: ComposerOwnership) => Promise<ComposerInput | null>, absent: (target: GuardTarget) => Promise<boolean>,
|
||||
activity?: (target: GuardTarget) => Promise<number | null>): void {
|
||||
this.inspectInput = inspect;
|
||||
this.inputAbsent = absent;
|
||||
this.inputActivity = activity;
|
||||
}
|
||||
|
||||
/** Startup-only, before exposing routes or starting writers. A stopped operation cannot
|
||||
@@ -483,6 +497,9 @@ export class SeatDeliveryGuard {
|
||||
} else if (input.state !== "empty") {
|
||||
throw new DeliveryGuardError("draft_input_busy", "The seat has an unsubmitted input. Automatic delivery is held.");
|
||||
} else {
|
||||
// Right before the paste (the adapter's check passes the text): the read above is a snapshot, and a draft typed
|
||||
// after it would get the paste appended and then lose the Enter, with the paste left in the person's input (#3076).
|
||||
if (pasteText !== undefined) input = await this.quietBeforePaste(target, read, unknown);
|
||||
const uncertainty = pasteText === undefined ? null : composerPasteUncertainty(input, pasteText);
|
||||
if (uncertainty) {
|
||||
const message = uncertainty === "controls"
|
||||
@@ -496,6 +513,30 @@ export class SeatDeliveryGuard {
|
||||
}
|
||||
}
|
||||
|
||||
/** No recent writable-client activity before a short wait, the input still empty when read again after it, and still
|
||||
* no recent activity after that read. A time tmux can't report holds the paste, as an unreadable input does. The
|
||||
* read it returns is the one the paste goes on. Typing after this final check and before the paste can still leave
|
||||
* the notice beside the person's draft: the Enter is then refused, nothing is submitted or deleted (#3076). */
|
||||
private async quietBeforePaste(target: GuardTarget, read: () => Promise<ComposerInput | null>,
|
||||
unknown: (reading: ComposerInput | null) => DeliveryGuardError): Promise<ComposerInput> {
|
||||
const quiet = async () => {
|
||||
if (!this.inputActivity) return;
|
||||
const last = await this.inputActivity(target).then(at => ({ at }), () => null);
|
||||
if (!last) throw new DeliveryGuardError("draft_input_unknown", "OpenRig couldn't read the recent activity of the seat's terminal clients. Automatic delivery is held; nothing was pasted.");
|
||||
// tmux reports whole seconds: activity at N.999 s reads as N, so the quiet window counts from the end of that second.
|
||||
if (last.at !== null && Date.now() - (last.at + CLIENT_ACTIVITY_PRECISION_MS) < INPUT_QUIET_MS) {
|
||||
throw new DeliveryGuardError("draft_input_busy", "A writable client of this seat had recent activity. Automatic delivery is held; nothing was pasted.");
|
||||
}
|
||||
};
|
||||
await quiet();
|
||||
await this.sleep(PASTE_SETTLE_MS);
|
||||
const again = await read();
|
||||
if (!again || again.state === "unknown") throw unknown(again);
|
||||
if (again.state !== "empty") throw new DeliveryGuardError("draft_input_busy", "The input changed just before the paste. Automatic delivery is held; nothing was pasted.");
|
||||
await quiet();
|
||||
return again;
|
||||
}
|
||||
|
||||
/** A leftover paste is OpenRig's only while it's exactly there: the same occupant, no person's input through OpenRig
|
||||
* since, and its text or the label recorded for it. A label first seen later proves nothing (#1871). */
|
||||
private ownsLeftover(target: GuardTarget, input: ComposerInput): boolean {
|
||||
|
||||
@@ -48,6 +48,8 @@ export interface SlowOperationInstrumentation {
|
||||
classify?: (value: T) => "ok" | "failed",
|
||||
): Promise<T>;
|
||||
recordRequest?(site: string, durationMs: number): void;
|
||||
/** A synchronous site's duration, recorded only at or above the slow threshold (runSyncSiteWhenSlow). */
|
||||
recordSlowSync?(site: string, durationMs: number, outcome: "ok" | "failed"): void;
|
||||
/** Bounded nonblocking diagnostic append; false means observation was dropped. */
|
||||
recordDiagnostic?(record: Record<string, unknown>): boolean;
|
||||
snapshot?(): SlowOperationSnapshot;
|
||||
@@ -262,6 +264,11 @@ export class SlowOpRecorder implements SlowOperationInstrumentation {
|
||||
this.recordMeasurement(`request:${site}`, durationMs);
|
||||
}
|
||||
|
||||
recordSlowSync(site: string, durationMs: number, outcome: "ok" | "failed" = "ok"): void {
|
||||
if (!Number.isFinite(durationMs) || durationMs < this.slowThresholdMs) return;
|
||||
this.recordMeasurement(site, durationMs, outcome);
|
||||
}
|
||||
|
||||
recordDiagnostic(record: Record<string, unknown>): boolean {
|
||||
this.diagnostic.offered++;
|
||||
const coverage = { ...this.diagnostic, pending: this.diagnosticIds.size,
|
||||
|
||||
@@ -10,6 +10,26 @@ export function runSyncSite<T>(site: string, fn: () => T): T {
|
||||
return recorder?.runSync ? recorder.runSync(site, fn) : fn();
|
||||
}
|
||||
|
||||
/**
|
||||
* runSyncSite for a site on a hot path (per request, per row), where a begin and an end record per call would crowd
|
||||
* the bounded log: it records only a call that took at least the recorder's slow threshold, and writes no begin
|
||||
* record, so a call that wedges the loop is named once it ends rather than while it runs.
|
||||
*/
|
||||
export function runSyncSiteWhenSlow<T>(site: string, fn: () => T): T {
|
||||
const record = recorder?.recordSlowSync?.bind(recorder);
|
||||
if (!record) return fn();
|
||||
const startedAt = performance.now();
|
||||
let outcome: "ok" | "failed" = "failed";
|
||||
try {
|
||||
const value = fn();
|
||||
outcome = "ok";
|
||||
return value;
|
||||
} finally {
|
||||
// Measurement only: a failing record never replaces fn's own result or error.
|
||||
try { record(site, performance.now() - startedAt, outcome); } catch { /* the recorder reports its own health */ }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Async twin of runSyncSite: instruments a NON-blocking site through the
|
||||
* recorder's async `runStage` seam (falls back to a bare call when no recorder
|
||||
|
||||
@@ -32,7 +32,7 @@ function signal() {
|
||||
return { ready: new Promise<void>(resolve => { release = resolve; }), release: () => release() };
|
||||
}
|
||||
|
||||
function fixture(saved?: Buffer, opts: { runtime?: string; idleHook?: boolean } = {}) {
|
||||
function fixture(saved?: Buffer, opts: { runtime?: string; idleHook?: boolean; appendPaste?: boolean } = {}) {
|
||||
const db = new Database(saved ?? ":memory:");
|
||||
if (!saved) {
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
@@ -42,23 +42,27 @@ function fixture(saved?: Buffer, opts: { runtime?: string; idleHook?: boolean }
|
||||
INSERT INTO sessions(id,node_id,session_name,status) VALUES ('sa','a','worker@test','running'),('sb','b','sibling@test','running');
|
||||
INSERT INTO occupant_tenures(id,node_id,generation_ordinal,generation_uuid,kind) VALUES ('ga','a',1,'g1','fresh'),('gb','b',1,'g2','fresh');`);
|
||||
}
|
||||
const guard = new SeatDeliveryGuard(db, name => resolveGuardTarget(db, name));
|
||||
const hooks: { load?: () => void | Promise<void>; paste?: () => void | Promise<void>; sleep?: (ms: number) => void | Promise<void>;
|
||||
settle?: (ms: number) => void | Promise<void>; clients?: () => string } = {};
|
||||
// The guard's own waits (its re-reads) go to `settle`, kept apart from the transport's `sleep`; none is real time.
|
||||
const guard = new SeatDeliveryGuard(db, name => resolveGuardTarget(db, name), async ms => { await hooks.settle?.(ms); });
|
||||
guard.recoverActivation();
|
||||
const bodies: Record<string, string | null> = { a: "", b: "" };
|
||||
const history: Record<string, string> = { a: "", b: "" };
|
||||
const writes: string[] = [], submissions: string[] = [];
|
||||
const observations: unknown[] = [];
|
||||
const hooks: { load?: () => void | Promise<void>; paste?: () => void | Promise<void>; sleep?: (ms: number) => void | Promise<void> } = {};
|
||||
let staged = "";
|
||||
// A terminal pane shows a submitted line as typed; an agent pane echoes a sent message under its prompt, as Claude does.
|
||||
const echo = (node: string, body: string) => (db.prepare("SELECT runtime FROM nodes WHERE id=?").get(node) as { runtime: string }).runtime === "terminal"
|
||||
? `${body}\n` : `${body.split("\n").map((row, i) => (i ? " " : "❯ ") + row).join("\n")}\n`;
|
||||
const tmux = new TmuxAdapter(async command => {
|
||||
if (command.includes("list-clients")) return hooks.clients?.() ?? "";
|
||||
if (command.includes("load-buffer")) await hooks.load?.();
|
||||
const node = command.includes("%2") ? "b" : "a";
|
||||
if (command.includes("paste-buffer")) {
|
||||
await hooks.paste?.();
|
||||
writes.push(`paste:${node}:${staged}`); bodies[node] = staged;
|
||||
// appendPaste: a paste lands after whatever the input already holds, as in a real composer (#3076 tests).
|
||||
writes.push(`paste:${node}:${staged}`); bodies[node] = opts.appendPaste ? (bodies[node] ?? "") + staged : staged;
|
||||
}
|
||||
if (command.includes("send-keys")) {
|
||||
writes.push(`keys:${node}:${command}`);
|
||||
@@ -1156,3 +1160,158 @@ describe("draft-aware at a full held quota (#2430)", () => {
|
||||
expect(f.outbox.retentionUsage("a").count).toBe(100);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("#3076: a draft typed between the guard's read and the paste", () => {
|
||||
const draft = "Please send it to the team. Also";
|
||||
const notice = "[OpenRig watchdog scheduler · policy: periodic-reminder · job: j]\nWake timer fired.";
|
||||
const pasted = (f: ReturnType<typeof fixture>) => f.writes.filter(w => w.startsWith("paste:"));
|
||||
const setActivity = (f: ReturnType<typeof fixture>, at: () => number | null) => {
|
||||
vi.spyOn(f.tmux, "lastClientActivity").mockImplementation(async () => at());
|
||||
};
|
||||
|
||||
it("is seen by the re-read after the settle wait: the notice is held and nothing is pasted into the person's input", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
const waits: number[] = [];
|
||||
// The person starts typing during the guard's wait right before the paste (the first read saw an empty input).
|
||||
f.hooks.settle = ms => { waits.push(ms); if (!pasted(f).length) f.bodies.a = draft; };
|
||||
const result = await f.send("watchdog-wake", notice, { openrigNotice: true, verify: false });
|
||||
expect(result).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
|
||||
expect(waits[0]).toBe(300);
|
||||
expect(pasted(f)).toEqual([]);
|
||||
expect(f.submissions).toEqual([]);
|
||||
expect(f.bodies.a).toBe(draft);
|
||||
});
|
||||
|
||||
it("holds on recent writable-client activity, also activity during the wait, and a later attempt delivers once quiet", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
let last: number | null = Date.now() - 1000;
|
||||
setActivity(f, () => last);
|
||||
const busy = await f.send("wake-1", "Wake timer fired.", { openrigNotice: true, verify: false });
|
||||
expect(busy).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
|
||||
|
||||
last = Date.now() - 60_000;
|
||||
f.hooks.settle = () => { last = Date.now(); }; // activity during the settle wait, nothing drawn yet
|
||||
f.advance();
|
||||
const during = await f.send("wake-2", "Wake timer fired again.", { openrigNotice: true, verify: false });
|
||||
expect(during).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
|
||||
expect(pasted(f)).toEqual([]);
|
||||
|
||||
last = Date.now() - 60_000;
|
||||
f.hooks.settle = undefined;
|
||||
f.advance();
|
||||
const sent = await f.send("wake-3", "Wake timer fired a third time.", { openrigNotice: true, verify: false });
|
||||
expect(sent).toMatchObject({ ok: true });
|
||||
expect(f.submissions).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("counts the quiet window from the end of the whole second tmux reports: 3.5 s ago still holds, 4.5 s ago delivers", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
let last = Date.now() - 3500; // activity at N.999 s reads as N: it may have been 2.5 s ago
|
||||
setActivity(f, () => last);
|
||||
const held = await f.send("whole-second", "Wake timer fired.", { openrigNotice: true, verify: false });
|
||||
expect(held).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
|
||||
last = Date.now() - 4500;
|
||||
f.advance();
|
||||
const sent = await f.send("whole-second-later", "Wake timer fired again.", { openrigNotice: true, verify: false });
|
||||
expect(sent).toMatchObject({ ok: true });
|
||||
});
|
||||
|
||||
it("holds as unreadable when tmux can't report recent writable-client activity; with no writable client it delivers", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
vi.spyOn(f.tmux, "lastClientActivity").mockRejectedValue(new Error("command timed out"));
|
||||
const held = await f.send("activity-unreadable", notice, { openrigNotice: true, verify: false });
|
||||
expect(held).toMatchObject({ sent: false, reason: "draft_input_unknown" });
|
||||
expect(pasted(f)).toEqual([]);
|
||||
setActivity(f, () => null);
|
||||
f.advance();
|
||||
const sent = await f.send("no-client", "Wake timer fired again.", { openrigNotice: true, verify: false });
|
||||
expect(sent).toMatchObject({ ok: true });
|
||||
});
|
||||
|
||||
it("holds as unreadable when the input can't be read after the wait", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
f.hooks.settle = () => { if (!pasted(f).length) f.bodies.a = null; };
|
||||
const result = await f.send("unreadable", notice, { openrigNotice: true, verify: false });
|
||||
expect(result).toMatchObject({ sent: false, reason: "draft_input_unknown" });
|
||||
expect(pasted(f)).toEqual([]);
|
||||
});
|
||||
|
||||
it("runs the send's own waiting checks before the guard's last read, so a draft typed while they wait is seen", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
// A conditional send's checks await (a conversation lookup); the person types meanwhile. The adapter returns the
|
||||
// guard's refusal as its result.
|
||||
const result = await f.guard.operation("worker@test", () =>
|
||||
f.tmux.sendText("worker@test", notice, undefined, { beforePaste: async () => { f.bodies.a = draft; } }));
|
||||
expect(result).toMatchObject({ ok: false, code: "draft_input_busy" });
|
||||
expect(pasted(f)).toEqual([]);
|
||||
expect(f.bodies.a).toBe(draft);
|
||||
});
|
||||
|
||||
it("applies to every automatic input under draft-aware, a deliberate send included, as the empty check already did", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
f.hooks.settle = () => { if (!pasted(f).length) f.bodies.a = draft; };
|
||||
const result = await f.send("deliberate", "a message from another seat", { verify: false });
|
||||
expect(result).toMatchObject({ sent: false, reason: "draft_input_busy" });
|
||||
expect(pasted(f)).toEqual([]);
|
||||
expect(f.bodies.a).toBe(draft);
|
||||
});
|
||||
|
||||
it("the remaining gap: typing after the final check and before the paste leaves the notice beside the draft, never submitted or deleted", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
let typed = false;
|
||||
// The person types between the guard's last check and tmux's paste: the paste lands after their text.
|
||||
f.hooks.paste = () => { if (!typed) { typed = true; f.bodies.a = draft; } };
|
||||
const result = await f.send("final-boundary", notice, { openrigNotice: true, verify: false });
|
||||
// Contamination: exactly one paste, and the input is the person's text followed by it, nothing removed.
|
||||
expect(pasted(f)).toHaveLength(1);
|
||||
const staged = pasted(f)[0]!.slice("paste:a:".length);
|
||||
expect(f.bodies.a).toBe(draft + staged);
|
||||
// Submission: none. No key at all reached the pane (no Enter, C-m or C-u), and the guard says why.
|
||||
expect(f.writes.filter(w => w.startsWith("keys:a"))).toEqual([]);
|
||||
expect(f.submissions).toEqual([]);
|
||||
expect(result).toMatchObject({ ok: false });
|
||||
expect(JSON.stringify(result)).toContain("draft_input_changed");
|
||||
});
|
||||
|
||||
it("checks the activity again after the final read: activity that appears at that read holds the paste", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
let last: number | null = Date.now() - 60_000;
|
||||
setActivity(f, () => last);
|
||||
const real = f.tmux.captureComposerSnapshot.getMockImplementation()!;
|
||||
let armed = false;
|
||||
f.hooks.settle = () => { armed = true; };
|
||||
// The composer still reads empty at the final read, but a writable client has activity right then.
|
||||
vi.spyOn(f.tmux, "captureComposerSnapshot").mockImplementation(async (name: string) => {
|
||||
const shot = await real(name);
|
||||
if (armed) { armed = false; last = Date.now(); }
|
||||
return shot;
|
||||
});
|
||||
const result = await f.send("activity-at-final-read", notice, { openrigNotice: true, verify: false });
|
||||
expect(result).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
|
||||
expect(pasted(f)).toEqual([]);
|
||||
});
|
||||
|
||||
it("reads tmux's real list-clients output end to end: an attach in this second holds, a quiet client delivers", async () => {
|
||||
const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
|
||||
await f.enable();
|
||||
const now = () => Math.floor(Date.now() / 1000);
|
||||
f.hooks.clients = () => `1|${now()}\n0|${now()}\n`; // a viewer tile and a writable client attached this second
|
||||
const held = await f.send("attach-now", notice, { openrigNotice: true, verify: false });
|
||||
expect(held).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
|
||||
expect(pasted(f)).toEqual([]);
|
||||
f.hooks.clients = () => `1|${now()}\n0|${now() - 10}\n`; // only the read-only tile is active
|
||||
f.advance();
|
||||
const sent = await f.send("attach-quiet", "Wake timer fired again.", { openrigNotice: true, verify: false });
|
||||
expect(sent).toMatchObject({ ok: true });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,69 @@
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
import type Database from "better-sqlite3";
|
||||
import { createDb } from "../src/db/connection.js";
|
||||
import { migrate } from "../src/db/migrate.js";
|
||||
import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
|
||||
import { EventBus } from "../src/domain/event-bus.js";
|
||||
import { QueueRepository, createLinkedSuccessorLookup } from "../src/domain/queue-repository.js";
|
||||
import { configureSyncSiteRecorder, runSyncSiteWhenSlow } from "../src/domain/sync-site-wrap.js";
|
||||
|
||||
// The queue's synchronous reads named in the slow-operation log, so an event-loop stall during one is attributed to it.
|
||||
// A fake recorder takes every call; the real recorder's slow threshold is pinned in slow-op-recorder.test.ts.
|
||||
describe("queue reads named in the slow-operation log", () => {
|
||||
let db: Database.Database;
|
||||
let repo: QueueRepository;
|
||||
const sync: string[] = [];
|
||||
const slow: Array<{ site: string; outcome: string }> = [];
|
||||
beforeEach(() => {
|
||||
db = createDb();
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
repo = new QueueRepository(db, new EventBus(db));
|
||||
configureSyncSiteRecorder({
|
||||
runSync<T>(site: string, fn: () => T): T { sync.push(site); return fn(); },
|
||||
recordSlowSync(site: string, _durationMs: number, outcome: "ok" | "failed"): void { slow.push({ site, outcome }); },
|
||||
});
|
||||
});
|
||||
afterEach(() => { configureSyncSiteRecorder(undefined); sync.length = 0; slow.length = 0; db.close(); });
|
||||
|
||||
it("names a handoff", async () => {
|
||||
const source = await repo.create({ sourceSession: "planner@r", destinationSession: "relay@r", body: "work", nudge: false });
|
||||
sync.length = 0;
|
||||
const { created } = await repo.handoff({ qitemId: source.qitemId, fromSession: "relay@r", toSession: "worker@r", nudge: false });
|
||||
expect(created.destinationSession).toBe("worker@r");
|
||||
expect(sync).toContain("queue.handoff");
|
||||
});
|
||||
|
||||
it("names decisionRevision's reply read through the slow-only site, once per revision", async () => {
|
||||
const ask = await repo.create({ sourceSession: "planner@r", destinationSession: "relay@r", body: "work", nudge: false });
|
||||
slow.length = 0;
|
||||
const revision = repo.decisionRevision(repo.getById(ask.qitemId)!);
|
||||
expect(typeof revision).toBe("string");
|
||||
expect(slow).toEqual([{ site: "queue.decision_revision.replies", outcome: "ok" }]);
|
||||
});
|
||||
|
||||
it("names the stuck sweep's linked-successor build when it is built", () => {
|
||||
const lookup = createLinkedSuccessorLookup(db, ["qitem-source"]);
|
||||
expect(sync).toEqual([]);
|
||||
expect(lookup.has("qitem-source", "worker@r")).toBe(false);
|
||||
expect(lookup.has("qitem-source", "other@r")).toBe(false);
|
||||
expect(sync).toEqual(["queue.linked_successors.build"]);
|
||||
});
|
||||
|
||||
it("keeps the wrapped call's result and error, and records each outcome", () => {
|
||||
expect(runSyncSiteWhenSlow("test.site", () => 7)).toBe(7);
|
||||
expect(() => runSyncSiteWhenSlow("test.site", () => { throw new Error("read failed"); })).toThrow("read failed");
|
||||
expect(slow).toEqual([{ site: "test.site", outcome: "ok" }, { site: "test.site", outcome: "failed" }]);
|
||||
});
|
||||
|
||||
it("never lets a failing record replace the wrapped call's result", () => {
|
||||
configureSyncSiteRecorder({ recordSlowSync(): void { throw new Error("recorder gone"); } });
|
||||
expect(runSyncSiteWhenSlow("test.site", () => "value")).toBe("value");
|
||||
expect(() => runSyncSiteWhenSlow("test.site", () => { throw new Error("read failed"); })).toThrow("read failed");
|
||||
});
|
||||
|
||||
it("runs the call bare when no recorder is configured", () => {
|
||||
configureSyncSiteRecorder(undefined);
|
||||
expect(runSyncSiteWhenSlow("test.site", () => "bare")).toBe("bare");
|
||||
expect(slow).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -21,6 +21,7 @@ const expectedSites = new Map<string, string[]>([
|
||||
["domain/codex-thread-id.ts", ["codex_thread_id.resolve_home"]],
|
||||
["domain/review/gather.ts", ["review.gather.git"]],
|
||||
["domain/tmux-option-defaults.ts", ["tmux_options.command_v"]],
|
||||
["domain/queue-repository.ts", ["queue.decision_revision.replies", "queue.linked_successors.build", "queue.handoff"]],
|
||||
]);
|
||||
|
||||
const pluginVendorUrl = "https://github.com/mvschwarz/openrig-plugins/releases/latest/download/openrig-core.tar.gz";
|
||||
@@ -290,6 +291,23 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
|
||||
await recorder.close();
|
||||
});
|
||||
|
||||
it("records a slow synchronous site only at or above the slow threshold, with its outcome", async () => {
|
||||
const mod = await loadRecorderModule();
|
||||
expect(mod, "slow-op-recorder production module is missing").not.toBeNull();
|
||||
if (!mod) return;
|
||||
|
||||
const dir = tempDir();
|
||||
const logPath = path.join(dir, "slow-operations.jsonl");
|
||||
const recorder = new mod.SlowOpRecorder({ logPath, slowThresholdMs: 250 });
|
||||
recorder.recordSlowSync("queue.decision_revision.replies", 249, "ok");
|
||||
recorder.recordSlowSync("queue.decision_revision.replies", 250, "ok");
|
||||
recorder.recordSlowSync("queue.decision_revision.replies", 400, "failed");
|
||||
await recorder.flush();
|
||||
const records = (await readRecords(logPath)).filter((record) => record.site === "queue.decision_revision.replies");
|
||||
expect(records.map((record) => [record.phase, record.durationMs, record.outcome])).toEqual([["end", 250, "ok"], ["end", 400, "failed"]]);
|
||||
await recorder.close();
|
||||
});
|
||||
|
||||
// The request-timing MIDDLEWARE CONTRACT, hermetic + deterministic. The prior "real server" form
|
||||
// ran the full daemon (event-loop monitor + a globalThis.fetch monkey-patch), which is structurally
|
||||
// ambient-fragile under the concurrent-workspace gate: it intermittently returned 200 for an unmatched
|
||||
|
||||
@@ -51,6 +51,47 @@ describe("TmuxAdapter.listSessionClientTerminals", () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe("TmuxAdapter.lastClientActivity (#3076)", () => {
|
||||
const activity = (stdout: string) => adapterFor(mockExec({ "list-clients": { stdout } })).lastClientActivity("op@kernel");
|
||||
|
||||
it("is the latest activity of a writable client, in milliseconds", async () => {
|
||||
expect(await activity("0|1790000100\n0|1790000105\n")).toBe(1790000105000);
|
||||
});
|
||||
|
||||
it("does not count a read-only client (a viewer tile)", async () => {
|
||||
expect(await activity("1|1790000200\n0|1790000100\n")).toBe(1790000100000);
|
||||
expect(await activity("1|1790000200\n")).toBeNull();
|
||||
});
|
||||
|
||||
it("counts an attach as recent activity: a key in the same second as the attach can't be told apart from it", async () => {
|
||||
// tmux would report client_created == client_activity for both an attach alone and an attach plus a key that second.
|
||||
expect(await activity("0|1790000300\n")).toBe(1790000300000);
|
||||
});
|
||||
|
||||
it("is null with no client", async () => {
|
||||
expect(await activity("")).toBeNull();
|
||||
expect(await activity("\n")).toBeNull();
|
||||
});
|
||||
|
||||
it("throws on a row it can't read, alone or next to a readable one, so the guard holds", async () => {
|
||||
for (const stdout of ["0|abc\n", "0|0\n", "0\n", "2|1790000100\n", "0|1790000000|1790000100\n", "0|1790000100\n0|abc\n"]) {
|
||||
await expect(activity(stdout)).rejects.toThrow();
|
||||
}
|
||||
});
|
||||
|
||||
it("throws when tmux can't say", async () => {
|
||||
for (const error of [new Error("can't find session: op@kernel"), NO_SERVER_ERROR, new Error("command timed out")]) {
|
||||
await expect(adapterFor(mockExec({ "list-clients": { error } })).lastClientActivity("op@kernel")).rejects.toThrow();
|
||||
}
|
||||
});
|
||||
|
||||
it("asks tmux for the read-only flag and the activity of each client of the session", async () => {
|
||||
const commands: string[] = [];
|
||||
await adapterFor(async command => { commands.push(command); return ""; }).lastClientActivity("op@kernel");
|
||||
expect(commands.some(c => c.includes("list-clients") && c.includes("#{client_readonly}|#{client_activity}"))).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("TmuxAdapter", () => {
|
||||
it("starts only an empty terminal server and reconciles repeated requests", async () => {
|
||||
let live = false;
|
||||
|
||||
@@ -158,6 +158,7 @@ async function driveEveryMethod(): Promise<RunCall[]> {
|
||||
await a.readPaneLastActivity(SESSION);
|
||||
await a.listClients();
|
||||
await a.listSessionClientTerminals(SESSION);
|
||||
await a.lastClientActivity(SESSION); // #3076: the client activity read right before an automatic paste
|
||||
await a.switchClient("/dev/ttys001", SESSION);
|
||||
// A runtime switch's exchange: two session-id reads, then the guarded if-shell swap (nested quoting); and its env writes.
|
||||
await a.swapRuntimeSwitchPanes({ nodeId: "n1", stagingSession: "staging'q a", canonicalSession: SESSION, canonicalPane: "%1", stagedPane: "%1" });
|
||||
|
||||
@@ -10,7 +10,7 @@ export function toConnectedAssistants(body: unknown): ConnectedAssistant[] | nul
|
||||
if (!Array.isArray(members)) return null;
|
||||
return members.flatMap((member: Record<string, unknown>) => {
|
||||
if (typeof member?.seat !== "string" || !member.seat) return [];
|
||||
const known = typeof member.status === "string" && member.status in CHECK_IN_DISPLAY;
|
||||
const known = typeof member.status === "string" && Object.hasOwn(CHECK_IN_DISPLAY, member.status);
|
||||
return [{ seat: member.seat, checkIn: {
|
||||
status: known ? member.status as CheckInStatus : "unknown",
|
||||
lastCheckInAt: typeof member.lastCheckInAt === "string" ? member.lastCheckInAt : null,
|
||||
|
||||
@@ -958,7 +958,11 @@ function contentLines(state: ViewState, snap: FleetSnapshot, contentWidth: numbe
|
||||
{ text: `Its seats aren't read in this view: rig ps --host ${rig.remoteHost} --nodes --rig ${rig.name}` },
|
||||
];
|
||||
if (rig.inventoryNotLoaded) return [{ text: `Reading ${rig.name}…` }];
|
||||
if (rig.inventoryUnavailable) return [{ text: `Inventory unavailable for ${rig.name} · refresh to Retry` }];
|
||||
if (rig.inventoryUnavailable) return [
|
||||
{ text: `Inventory unavailable for ${rig.name} · refresh to Retry` },
|
||||
...(leaf?.kind === "pod" ? [] : connectedAssistantLines(host.connectedAssistants, host.connectedAssistantsUnavailable,
|
||||
rig.name, state.filter, contentWidth, state.timeZone)),
|
||||
];
|
||||
const podFilter = leaf?.kind === "pod" ? leaf.name : null;
|
||||
const all = rig.pods.flatMap((p) => p.agents.map((a) => ({ pod: p.name, ...a })));
|
||||
const rows = all
|
||||
|
||||
@@ -782,6 +782,34 @@ describe("snapshot hydration over the §4.A reads (Phase 2)", () => {
|
||||
expect(instanceLines).not.toContain("not shown");
|
||||
});
|
||||
|
||||
it("shows inherited-property check-in statuses as unknown", async () => {
|
||||
const hosted = { members: ["constructor", "toString"].map((status, i) => ({
|
||||
seat: `guest${i}@myrig`, status, reason: "unrecognized reason",
|
||||
})) };
|
||||
const snap = await hydrateSnapshot(fixtureClient({}, { "/api/queue/hosted-members": hosted }));
|
||||
const view = createViewState({ instanceId: "t", getSnapshot: () => snap });
|
||||
view.dispatch({ type: "drill", resource: "rig", name: "myrig" });
|
||||
const lines = renderScreen(view.get(), snap, { cols: 160, rows: 60 }).lines.join("\n");
|
||||
expect(lines).toContain("guest0@myrig unknown");
|
||||
expect(lines).toContain("guest1@myrig unknown");
|
||||
expect(lines).not.toContain("unrecognized reason");
|
||||
});
|
||||
|
||||
it("shows connected check-ins beside a failed rig nodes read", async () => {
|
||||
const hosted = { members: [
|
||||
{ seat: "chat@myrig", status: "on-request", lastCheckInAt: null },
|
||||
{ seat: "other@downrig", status: "checking-in", lastCheckInAt: null },
|
||||
] };
|
||||
const snap = await hydrateSnapshot(fixtureClient({ "/api/rigs/01JRIG/nodes": { status: 500 } },
|
||||
{ "/api/queue/hosted-members": hosted }));
|
||||
const view = createViewState({ instanceId: "t", getSnapshot: () => snap });
|
||||
view.dispatch({ type: "drill", resource: "rig", name: "myrig" });
|
||||
const lines = renderScreen(view.get(), snap, { cols: 160, rows: 60 }).lines.join("\n");
|
||||
expect(lines).toContain("Inventory unavailable for myrig");
|
||||
expect(lines).toContain("chat@myrig on request:");
|
||||
expect(lines).not.toContain("other@downrig");
|
||||
});
|
||||
|
||||
it("a known-empty list of connected assistants stays quiet", async () => {
|
||||
const snap = await hydrateSnapshot(fixtureClient());
|
||||
expect(snap.hosts[0]!.connectedAssistants).toEqual([]);
|
||||
|
||||
@@ -69,7 +69,7 @@
|
||||
"agent-operated-software/SKILL.md": "ec910bf5f631d7f20872786a565a0437a446fa2d5fd8b52b685979978243a6a8",
|
||||
"agent-operated-workflows/SKILL.md": "24e54cbfa979999f523516112670618b2926ae5d42c7aa0846c94db214857ea5",
|
||||
"applying-a-permission-policy/SKILL.md": "f7bba8d50b10fc8715f40a57bf82677ca9388c6718c8a162833d9c6daf2afdca",
|
||||
"claude-compaction-restore/SKILL.md": "79a707974824e91d5888c7bc10b152989ff891491f1adecf3e93991e2af55f81",
|
||||
"claude-compaction-restore/SKILL.md": "cd8369a62647ad389a9b0cc91c6f784a0d93e2d2b874bf9570d47c2134647d1c",
|
||||
"claude-compaction-restore/reference/SKILL-v1.md": "6af2a889c9c38227fb186897c6e29185f81b215822070b469a6ceeb47734e133",
|
||||
"claude-compaction-restore/scripts/precompact-hook.mjs": "712e97c8beb02eb6c2815884a960778a225d8f6238dfe04a9b0547c34eaa131d",
|
||||
"claude-compaction-restore/scripts/restore-from-jsonl.mjs": "5584cff4ab36cd843e14c81bd201e4a2f8e498e995abacc4f81ec6c1fc556b10",
|
||||
|
||||
Reference in new issue
Block a user