Compare commits

...
Author SHA1 Message Date
Mike Schwarz b1460a11d9 docs(skills): after a compaction, a hold covers only its own work (#3111)
A Claude seat restoring after a compaction reads a hold as governing only the work it covers. Three lines in the claude-compaction-restore skill change. A hold you verified as live, a release order or an operator's own restore map overrides the default order for the work it covers, and the seat does the unaffected work. The reconnect note waits on a live hold that covers its next step. The read-depth table and the reconnect note are written as the seat resumes. "Rebuild the connections before you act on anything" is unchanged, so reading still comes first, and a live hold still stops the work it covers. The managed post-restore audit message still asks for the table before other work; when it arrives, it is the live instruction. The skill's digest is regenerated.
2026-11-12 19:41:07 -08:00
Mike Schwarz d4855cbbec docs: scope two promises to what they cover (#3106)
Two sentences now say exactly what they cover. The startup guide's managed-block line reads "tidying the copies never removes a text you wrote": only a copy that repeats a kept one, or OpenRig's own copy beside one you edited, is taken out, while a launch can still replace a block OpenRig has no record of writing when it differs from its source. rig down's help says that for a community harness only its pane stops; its own processes may keep running, and rig down names it. Text only, with no behaviour change.
2026-11-12 19:39:11 -08:00
Mike Schwarz b5df49407c perf(daemon): name the queue's synchronous reads in the slow-operation log (#3100)
When the daemon's event loop stalls, the slow-operation log can now name three more of the queue's synchronous reads as the cause: a whole handoff (queue.handoff), the stuck sweep's linked-successor read (queue.linked_successors.build), and the reply read behind each row's decision revision (queue.decision_revision.replies). The first two record a begin and an end. The third records only a call at or above the recorder's existing slow threshold, through a new runSyncSiteWhenSlow, so the bounded log isn't crowded; such a call is named once it ends, not while it runs.

This changes no behaviour: a failing record never replaces the wrapped call's own result or error. It adds evidence for the next diagnosis, not a cause; event-loop gaps that match no logged operation stay unattributed.
2026-11-12 19:01:39 -08:00
billbenetti f46bd7ec08 fix(daemon): re-check a seat's input right before an automatic paste (#3076) (#3097)
A mitigation of the delivery race in #3076: it narrows the window in which an automatic notice can be pasted into a draft a person is typing. Before, the draft-aware guard read the seat's input once, found it empty, and pasted; a person who started typing in between got the notice appended to their draft. Now, right before an automatic paste, the send's own waiting checks run first, then the guard holds when a writable client had activity in the last 3 s (session-wide tmux client activity, not proof of typing in this pane), re-reads the input after a 300 ms wait and holds unless it's still empty, and holds as draft_input_unknown when tmux can't be read. A held paste uses the existing retained handling, so a later attempt delivers once the input reads quiet and empty. An ordinary guarded paste now waits about 300 ms longer.

Typing after the guard's final check and before the paste is still not seen. The notice then lands beside the draft, the Enter is refused, nothing is submitted and nothing is deleted; a test pins that outcome. #3076 stays open for that window.

Thanks to billbenetti.
2026-11-12 18:39:26 -08:00
Mike Schwarz c411603c8c docs(changelog): 0.7.13's upgrade steps name the install (#3099)
0.7.13's "Before you upgrade" now names every step in order. It said to stop the daemon, back up the database and start it again, but never when to install the new version. The bullet now reads: back up the database, install 0.7.13, then restart the daemon, with rig daemon stop, a copy of the database file (rig config get db.path prints its path), installing OpenRig 0.7.13 with your existing installation method, and then rig daemon start. Text only.
2026-11-12 18:30:34 -08:00
Mike Schwarz 39f83a7904 fix(tui): retain connected check-ins when inventory fails (#3105)
Connected assistants keep their check-in lines visible in the TUI when a rig's seat inventory can't be read; the inventory error stays visible beside them. An unrecognized check-in status reads unknown, including names that are inherited JavaScript properties, since only a status's own keys are checked. It reuses the existing assistant renderer and adds no requests or dependencies.
2026-11-12 18:25:52 -08:00
18 changed files with 445 additions and 25 deletions

No files matched your search

+3 -3
View File
@@ -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))
+1 -1
View File
@@ -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.
+2 -1
View File
@@ -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
@@ -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
+28 -4
View File
@@ -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;
+11 -4
View File
@@ -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
+41
View File
@@ -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" });
+1 -1
View File
@@ -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,
+5 -1
View File
@@ -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
+28
View File
@@ -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([]);
+1 -1
View File
@@ -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",