mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-10 17:30:37 -10:00
Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
c2226648c4 | ||
|
|
4be95ef23f | ||
|
|
c6086a44ab |
No files matched your search
@@ -428,24 +428,22 @@ running seat only through the action in the last column.
|
||||
|
||||
### New guidance for a running seat
|
||||
|
||||
There are three ways for a seat to pick up new guidance:
|
||||
These reach a running seat without a restart, in the order they reach the agent:
|
||||
|
||||
1. **Tell it.** `rig send <seat> "Read <file>: <why>"`, or `rig walk <seat> --through <files>` for several. The seat's
|
||||
interruption settings can hold a message; `rig send` may report `retained, not delivered`.
|
||||
1. **Tell it.** `rig send <seat> "Read <file>: <why>"`, or `rig walk <seat> --through <files>` for several. It's in the
|
||||
conversation at the seat's next turn. The seat's reply is the only evidence that it read the file.
|
||||
2. **The refocus hook**, where the seat runs OpenRig's `openrig-core` hooks (**Refocus** in `rig seat status` shows whether
|
||||
the hook has recorded it). The hook's content comes from `OPENRIG_REFOCUS_CONTENT_REF` (a `rig context get` address),
|
||||
else `OPENRIG_REFOCUS_CONTENT_FILE`, else `refocus/REFOCUS.md` under `OPENRIG_HOME`, else the shipped default. After
|
||||
compaction the hook waits for an eligible prompt. For managed Claude compaction it skips the acknowledgement boundary,
|
||||
and other prompts can arrive first and be skipped while restore is pending. Claude can also refocus after the
|
||||
conversation grows past the hook's size threshold.
|
||||
else `OPENRIG_REFOCUS_CONTENT_FILE`, else `refocus/REFOCUS.md` under `OPENRIG_HOME`, else the shipped default. It
|
||||
reaches the agent at the next prompt after a compaction, and for Claude also once the conversation has grown past the
|
||||
hook's size threshold.
|
||||
3. **Its guidance file, loaded again:** for Claude at its next compaction, `/clear` or start, for Codex at its next start.
|
||||
**Guidance** in `rig seat status` shows which of OpenRig's blocks were written after the session's recorded startup finish.
|
||||
|
||||
A file pointer leaves the read to the seat, under its own permission settings. `rig walk` reads the named files in the
|
||||
calling process and sends their contents. A seat whose harness asks before reading a file outside its project (its own
|
||||
`LEARNED.md` under the topology root, a plugin's skill) waits at that prompt until someone answers it, even though a hook
|
||||
could deliver an excerpt of the same file. Finding a file, a hook picking it up, and the agent reading and using it are
|
||||
separate facts.
|
||||
A pointer or a message names a file; the seat still reads it itself, under its own permission settings. A seat whose
|
||||
harness asks before reading a file outside its project (its own `LEARNED.md` under the topology root, a plugin's skill)
|
||||
waits at that prompt until someone answers it, even though a hook could deliver an excerpt of the same file. Finding a
|
||||
file, a hook picking up, and the agent reading and using it are separate facts.
|
||||
|
||||
Per harness: Claude Code's and Codex's permission choices are in the setup reference
|
||||
([Claude Code](getting-started.md#claude-code-a-different-launch-flag),
|
||||
|
||||
@@ -64,6 +64,20 @@ function conversationFields(providerPayload, env) {
|
||||
};
|
||||
}
|
||||
|
||||
// The prompt the harness received at UserPromptSubmit, as a fingerprint and byte length, with Claude's own prompt id
|
||||
// when it sends one: enough for the daemon to match it to a message OpenRig wrote, without the prompt leaving this
|
||||
// process. Any other event, or a prompt that isn't a string, adds nothing.
|
||||
function promptBoundaryFields(hookEvent, providerPayload) {
|
||||
if (hookEvent !== "UserPromptSubmit" || typeof providerPayload.prompt !== "string") return {};
|
||||
const bytes = Buffer.from(providerPayload.prompt, "utf8");
|
||||
const nativePromptId = firstString(providerPayload.prompt_id);
|
||||
return {
|
||||
promptSha256: crypto.createHash("sha256").update(bytes).digest("hex"),
|
||||
promptBytes: bytes.length,
|
||||
...(nativePromptId && /^[A-Za-z0-9_-]{1,128}$/.test(nativePromptId) ? { nativePromptId } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
// The route this run reports by, so the daemon can count one event once per route: "plugin" for the plugin's hook
|
||||
// (started with --from-plugin), "managed" for the settings or managed hook copy.
|
||||
function buildOpenRigPayload(providerPayload, env = process.env, now = () => new Date(), argv = process.argv) {
|
||||
@@ -127,6 +141,7 @@ function buildOpenRigPayload(providerPayload, env = process.env, now = () => new
|
||||
&& (providerPayload.agent_id == null || firstString(providerPayload.agent_id))
|
||||
? { hookAgentId: firstString(providerPayload.agent_id) } : {}),
|
||||
turnId: firstString(providerPayload.turn_id, providerPayload.turnId),
|
||||
...promptBoundaryFields(hookEvent, providerPayload),
|
||||
occurredAt: now().toISOString(),
|
||||
relayRoute: argv.includes("--from-plugin") ? "plugin" : "managed",
|
||||
...conversationFields(providerPayload, env),
|
||||
|
||||
@@ -79,7 +79,7 @@ export class DraftAwareDelivery {
|
||||
// delivery needs an idle seat: submitting it starts a turn, so that delivery is held as before.
|
||||
await this.guard.beforeInput(session, false, undefined, opts.waitForIdleMs === undefined && !opts.preserveInput ? "recognize" : undefined);
|
||||
result = await this.deliver(session, text, { ...opts,
|
||||
onInputEffect: phase => { inputWritten = true; opts.onInputEffect?.(phase); },
|
||||
onInputEffect: (phase, written) => { inputWritten = true; opts.onInputEffect?.(phase, written); },
|
||||
beforeWrite: () => {
|
||||
if (opts.queueWake && !this.wakeApplicable?.(entries)) throw new DeliveryGuardError("draft_wake_superseded", "The queue wake is no longer applicable; no further input was written.");
|
||||
opts.beforeWrite?.();
|
||||
|
||||
@@ -28,6 +28,11 @@ interface ActiveEnvelope {
|
||||
registered: Set<PersistedEvent>;
|
||||
}
|
||||
|
||||
/** The SQL condition leaving `omit` out, bound as parameters: empty when there's nothing to leave out. */
|
||||
function typesLeftOut(omit: readonly string[]): string {
|
||||
return omit.length ? ` AND type NOT IN (${omit.map(() => "?").join(", ")})` : "";
|
||||
}
|
||||
|
||||
export class EventBus {
|
||||
private subscribers = new Set<Subscriber>();
|
||||
private activeEnvelope: ActiveEnvelope | null = null;
|
||||
@@ -292,28 +297,30 @@ export class EventBus {
|
||||
return rows.map((row) => this.rowToPersistedEvent(row));
|
||||
}
|
||||
|
||||
private pageStatements?: { all: Database.Statement; rig: Database.Statement };
|
||||
private pageStatements = new Map<string, Database.Statement>();
|
||||
|
||||
/** Up to `limit` events after `seq` and up to `upTo`, oldest first, of one rig when `rigId` is given: a replay read a
|
||||
* page at a time, which events stored meanwhile never lengthen. */
|
||||
replayPage(seq: number, upTo: number, limit: number, rigId?: string): PersistedEvent[] {
|
||||
this.pageStatements ??= {
|
||||
all: this.db.prepare("SELECT seq, rig_id, node_id, type, payload, created_at FROM events WHERE seq > ? AND seq <= ? ORDER BY seq LIMIT ?"),
|
||||
rig: this.db.prepare("SELECT seq, rig_id, node_id, type, payload, created_at FROM events WHERE rig_id = ? AND seq > ? AND seq <= ? ORDER BY seq LIMIT ?"),
|
||||
};
|
||||
const rows = (rigId ? this.pageStatements.rig.all(rigId, seq, upTo, limit) : this.pageStatements.all.all(seq, upTo, limit)) as EventRow[];
|
||||
/** Up to `limit` events after `seq` and up to `upTo`, oldest first, of one rig when `rigId` is given, leaving out the
|
||||
* `omit` types: a replay read a page at a time, which events stored meanwhile never lengthen. */
|
||||
replayPage(seq: number, upTo: number, limit: number, rigId?: string, omit: readonly string[] = []): PersistedEvent[] {
|
||||
const key = `${rigId ? "rig" : "all"}:${omit.length}`;
|
||||
let statement = this.pageStatements.get(key);
|
||||
if (!statement) {
|
||||
statement = this.db.prepare(`SELECT seq, rig_id, node_id, type, payload, created_at FROM events WHERE ${rigId ? "rig_id = ? AND " : ""}`
|
||||
+ `seq > ? AND seq <= ?${typesLeftOut(omit)} ORDER BY seq LIMIT ?`);
|
||||
this.pageStatements.set(key, statement);
|
||||
}
|
||||
const rows = statement.all(...(rigId ? [rigId] : []), seq, upTo, ...omit, limit) as EventRow[];
|
||||
return rows.map((row) => this.rowToPersistedEvent(row));
|
||||
}
|
||||
|
||||
/** The seq after which the newest `count` events start (of one rig when `rigId` is given): 0 when there are no more
|
||||
* than `count`, the newest seq for 0. A count past a safe integer is more than any log holds: never an OFFSET SQLite
|
||||
* cannot take. */
|
||||
seqBeforeNewest(count: number, rigId?: string): number {
|
||||
/** The seq after which the newest `count` events start (of one rig when `rigId` is given), not counting the `omit`
|
||||
* types: 0 when there are no more than `count`, the newest seq for 0. A count past a safe integer is more than any log
|
||||
* holds: never an OFFSET SQLite cannot take. */
|
||||
seqBeforeNewest(count: number, rigId?: string, omit: readonly string[] = []): number {
|
||||
if (count === 0) return this.maxSeq();
|
||||
if (!Number.isSafeInteger(count)) return 0;
|
||||
const row = (rigId
|
||||
? this.db.prepare("SELECT seq FROM events WHERE rig_id = ? ORDER BY seq DESC LIMIT 1 OFFSET ?").get(rigId, count)
|
||||
: this.db.prepare("SELECT seq FROM events ORDER BY seq DESC LIMIT 1 OFFSET ?").get(count)) as { seq: number } | undefined;
|
||||
const row = this.db.prepare(`SELECT seq FROM events WHERE ${rigId ? "rig_id = ?" : "1"}${typesLeftOut(omit)} ORDER BY seq DESC LIMIT 1 OFFSET ?`)
|
||||
.get(...(rigId ? [rigId] : []), ...omit, count) as { seq: number } | undefined;
|
||||
return row?.seq ?? 0;
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
import type { EventBus } from "./event-bus.js";
|
||||
import type { PersistedEvent } from "./types.js";
|
||||
|
||||
// A harness's own prompt hook (UserPromptSubmit) says what its input took; a delivery-evidence record says what OpenRig
|
||||
// wrote. Neither carries an OpenRig message id, so the two are matched only as a candidate join on the written text: the
|
||||
// same session and occupant generation, an equal fingerprint and byte length, the write no later than the hook, and
|
||||
// exactly one candidate on the other side. A match records that message's input boundary as observed. Anything else
|
||||
// (no candidate, two, a different generation, a stale entry, a daemon that restarted between the two) records nothing,
|
||||
// so the boundary stays unknown. Either side can arrive first: the hook may reach the daemon before the send returns.
|
||||
|
||||
/** One written message, as its delivery-evidence record gave it. */
|
||||
export interface BoundaryWrite { sessionName: string; generation: string; sha256: string; bytes: number; writtenAt: string; messageId: string; attemptId: string }
|
||||
/** One prompt hook's facts, as the relay sent them. */
|
||||
export interface BoundaryHook {
|
||||
sessionName: string; generation: string; sha256: string; bytes: number; occurredAt: string;
|
||||
runtime: string | null; nativePromptId: string | null; nativeTurnId: string | null;
|
||||
}
|
||||
|
||||
const PER_SESSION = 20;
|
||||
const WINDOW_MS = 10 * 60_000;
|
||||
|
||||
type Entry<T> = T & { at: number; matched: boolean };
|
||||
|
||||
export class InputBoundaryCorrelator {
|
||||
private writes = new Map<string, Entry<BoundaryWrite>[]>();
|
||||
private hooks = new Map<string, Entry<BoundaryHook>[]>();
|
||||
|
||||
constructor(private readonly eventBus: EventBus, private readonly now: () => number = Date.now) {}
|
||||
|
||||
/** A delivery-evidence record. Only a write whose text reached the input, with its generation known, can match. */
|
||||
noteRecord(event: PersistedEvent): void {
|
||||
if (event.type !== "transport.delivery_evidence") return;
|
||||
if (event.written !== "submitted" && event.written !== "text_submit_unknown") return;
|
||||
if (!event.occupantGeneration || !event.wireSha256 || event.wireBytes === null || !event.writtenAt) return;
|
||||
const write: BoundaryWrite = { sessionName: event.sessionName, generation: event.occupantGeneration, sha256: event.wireSha256,
|
||||
bytes: event.wireBytes, writtenAt: event.writtenAt, messageId: event.messageId, attemptId: event.attemptId };
|
||||
const hook = this.single(this.hooks, write.sessionName, (h) => same(write, h) && write.writtenAt <= h.occurredAt);
|
||||
if (hook) this.record(write, hook);
|
||||
else this.keep(this.writes, write);
|
||||
}
|
||||
|
||||
/** A prompt hook. */
|
||||
noteHook(hook: BoundaryHook): void {
|
||||
const write = this.single(this.writes, hook.sessionName, (w) => same(w, hook) && w.writtenAt <= hook.occurredAt);
|
||||
if (write) this.record(write, hook);
|
||||
else this.keep(this.hooks, hook);
|
||||
}
|
||||
|
||||
/** The one unmatched entry that fits, marked matched; null for none or more than one. */
|
||||
private single<T>(index: Map<string, Entry<T>[]>, sessionName: string, fits: (entry: T) => boolean): Entry<T> | null {
|
||||
const live = this.prune(index, sessionName);
|
||||
const candidates = live.filter((entry) => !entry.matched && fits(entry));
|
||||
if (candidates.length !== 1) return null;
|
||||
candidates[0]!.matched = true;
|
||||
return candidates[0]!;
|
||||
}
|
||||
|
||||
private keep<T>(index: Map<string, Entry<T>[]>, value: T & { sessionName: string }): void {
|
||||
const live = this.prune(index, value.sessionName);
|
||||
live.push({ ...value, at: this.now(), matched: false });
|
||||
index.set(value.sessionName, live.slice(-PER_SESSION));
|
||||
}
|
||||
|
||||
private prune<T>(index: Map<string, Entry<T>[]>, sessionName: string): Entry<T>[] {
|
||||
const cutoff = this.now() - WINDOW_MS;
|
||||
const live = (index.get(sessionName) ?? []).filter((entry) => entry.at >= cutoff);
|
||||
index.set(sessionName, live);
|
||||
return live;
|
||||
}
|
||||
|
||||
private record(write: BoundaryWrite, hook: BoundaryHook): void {
|
||||
try {
|
||||
this.eventBus.emit({
|
||||
type: "transport.input_boundary", sessionName: write.sessionName, occupantGeneration: write.generation,
|
||||
messageId: write.messageId, attemptId: write.attemptId, observed: "UserPromptSubmit", runtime: hook.runtime,
|
||||
nativePromptId: hook.nativePromptId, nativeTurnId: hook.nativeTurnId, observedAt: hook.occurredAt,
|
||||
});
|
||||
} catch { /* Best effort: a record that can't be written leaves the boundary unknown. */ }
|
||||
}
|
||||
}
|
||||
|
||||
function same(write: { generation: string; sha256: string; bytes: number }, hook: { generation: string; sha256: string; bytes: number }): boolean {
|
||||
return write.generation === hook.generation && write.sha256 === hook.sha256 && write.bytes === hook.bytes;
|
||||
}
|
||||
|
||||
const correlators = new WeakMap<EventBus, InputBoundaryCorrelator>();
|
||||
|
||||
/** The correlator for this event bus, created and subscribed on first use (startup creates it, so no write is missed). */
|
||||
export function inputBoundaryCorrelatorFor(eventBus: EventBus): InputBoundaryCorrelator {
|
||||
let correlator = correlators.get(eventBus);
|
||||
if (!correlator) {
|
||||
const created = new InputBoundaryCorrelator(eventBus);
|
||||
eventBus.subscribe((event) => created.noteRecord(event));
|
||||
correlators.set(eventBus, created);
|
||||
correlator = created;
|
||||
}
|
||||
return correlator;
|
||||
}
|
||||
@@ -116,6 +116,10 @@ export const PRUNABLE_EVENT_TYPES = [
|
||||
"queue.claimed",
|
||||
"queue.unclaimed",
|
||||
"queue.handed_off",
|
||||
// One delivery-evidence record per send, and at most one input-boundary record per send; kept only as long as the
|
||||
// other high-volume event types.
|
||||
"transport.delivery_evidence",
|
||||
"transport.input_boundary",
|
||||
] as const;
|
||||
|
||||
export interface RetentionOptions {
|
||||
|
||||
@@ -6,7 +6,7 @@ import { currentClaudeThemeMenu } from "./claude-theme-menu.js";
|
||||
import { currentClaudeResumePicker } from "./native-resume-probe.js";
|
||||
import stringWidth from "string-width";
|
||||
import { CODEX_LIVE_WORKING_FRAME_PATTERN, CODEX_TURN_STATUS_PATTERN } from "./codex-pane-patterns.js";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { CLAUDE_NOT_STATUS_LINE_ROW, COLLAPSED_PASTE, composerRegion, isComposerFooter, isClaudePasteHint, inspectStartupStagedText, QUEUED_PLACEHOLDER, startupSubmissionEvidence, type StartupSubmissionEvidence } from "./startup-submission-evidence.js";
|
||||
import {
|
||||
composerContainsOwnedText,
|
||||
@@ -156,6 +156,36 @@ export interface PaneActivityClassification {
|
||||
evidence: string | null;
|
||||
}
|
||||
|
||||
/** Refusal codes from the checks that run before Enter reaches the adapter (the seat's input guard, the expected
|
||||
* conversation and managed compaction): a written send that stopped with one left its text without Enter. */
|
||||
/** The written text's fingerprint, byte length and paste time, to match a harness's own prompt hook against; nulls when
|
||||
* nothing was pasted. The text itself is never recorded. */
|
||||
function wireFingerprint(wire: { text: string; at: string } | null): { wireSha256: string | null; wireBytes: number | null; writtenAt: string | null } {
|
||||
if (wire === null) return { wireSha256: null, wireBytes: null, writtenAt: null };
|
||||
const bytes = Buffer.from(wire.text, "utf8");
|
||||
return { wireSha256: createHash("sha256").update(bytes).digest("hex"), wireBytes: bytes.length, writtenAt: wire.at };
|
||||
}
|
||||
|
||||
function enterWithheld(reason: string | undefined): boolean {
|
||||
return typeof reason === "string" && (reason.startsWith("expected_identity_") || reason.startsWith("draft_") || reason === "managed_compaction_pending");
|
||||
}
|
||||
|
||||
/** What a send's own result shows reached the seat's input. A send that succeeded without being retained issued the text
|
||||
* and Enter, whether or not it checked the screen, except a bare Enter and a prompt answer typed without Enter. A
|
||||
* write that may have landed is uncertain. Text that was written reads text only when a pre-Enter check refused the
|
||||
* Enter, and otherwise text with the submission unknown, since Enter may have been attempted. A send that wrote
|
||||
* nothing says so. Anything else the result doesn't establish is unknown, never a guess. */
|
||||
export function writtenFact(result: SendResult): "submitted" | "submit_only" | "text_only" | "text_submit_unknown" | "uncertain" | "none" | "unknown" {
|
||||
if (result.ok && result.outcome !== "retained") {
|
||||
if (result.submitOnly) return "submit_only";
|
||||
return result.promptInteraction === "unverified" ? "text_only" : "submitted";
|
||||
}
|
||||
if (result.inputPhase === "attempted") return "uncertain";
|
||||
if (result.inputPhase === "written") return enterWithheld(result.reason) ? "text_only" : "text_submit_unknown";
|
||||
if (result.inputPhase === "not_reached" || result.sent === false) return "none";
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
function trimPaneLines(paneContent: string): string[] {
|
||||
return paneContent
|
||||
.split("\n")
|
||||
@@ -449,18 +479,14 @@ function closesFramedPromptAbove(lines: string[], open: number, indent: number,
|
||||
const codexFooter = (line: string): boolean => isComposerFooter(line)
|
||||
|| /(?:\b\d+% context left\b|\bContext \[)/.test(line);
|
||||
|
||||
/** Preserve the styling used for autocomplete ghosts and the drawn cursor.
|
||||
* Color channel values must never be interpreted as style codes. */
|
||||
/** Preserve faint styling so autocomplete ghosts can be separated from authored
|
||||
* text. Color channel values must never be interpreted as style codes. */
|
||||
function styledTextLines(screen: string) {
|
||||
let faint = false, inverse = false;
|
||||
let faint = false;
|
||||
return screen.replace(/\r\n/g, "\n").split("\n").map(raw => {
|
||||
let text = "", offset = 0;
|
||||
const faintCells: boolean[] = [], inverseCells: boolean[] = [];
|
||||
const append = (value: string) => {
|
||||
text += value;
|
||||
faintCells.push(...Array<boolean>(value.length).fill(faint));
|
||||
inverseCells.push(...Array<boolean>(value.length).fill(inverse));
|
||||
};
|
||||
const faintCells: boolean[] = [];
|
||||
const append = (value: string) => { text += value; faintCells.push(...Array<boolean>(value.length).fill(faint)); };
|
||||
for (const match of raw.matchAll(/\x1b\[([0-9;:]*)m/g)) {
|
||||
append(raw.slice(offset, match.index));
|
||||
const codes = match[1]!.split(";");
|
||||
@@ -471,18 +497,15 @@ function styledTextLines(screen: string) {
|
||||
const skip = codes[i + 1] === "5" ? 2 : codes[i + 1] === "2" ? (codes[i + 2] === "" ? 5 : 4) : 0;
|
||||
// Color channels are not style codes. Ambiguous compound sequences
|
||||
// provide no faint-style proof, rather than interpreting an RGB 2 as dim.
|
||||
if (!skip || i + skip !== codes.length - 1) { faint = inverse = false; break; }
|
||||
if (!skip || i + skip !== codes.length - 1) { faint = false; break; }
|
||||
i += skip;
|
||||
} else if (code === "" || code === "0") faint = inverse = false;
|
||||
else if (code === "22") faint = false;
|
||||
} else if (code === "" || code === "0" || code === "22") faint = false;
|
||||
else if (code === "2") faint = true;
|
||||
else if (code === "7") inverse = true;
|
||||
else if (code === "27") inverse = false;
|
||||
}
|
||||
offset = match.index! + match[0].length;
|
||||
}
|
||||
append(raw.slice(offset));
|
||||
return { text, faintCells, inverseCells };
|
||||
return { text, faintCells };
|
||||
});
|
||||
}
|
||||
|
||||
@@ -577,11 +600,7 @@ export function inspectComposerInput(snapshot: ComposerSnapshot | null, runtime:
|
||||
else if (cursorOffset >= 0 && !rows[cursorRow]!.slice(cursorOffset).trim()) visible[cursorRow] = rows[cursorRow]!.slice(0, cursorOffset);
|
||||
}
|
||||
const atStart = y === start && x === prefix;
|
||||
// Claude can draw an inverse-video blank in the cursor cell (#2925). Only
|
||||
// that single cell at the start is padding; ordinary/preceding spaces stay input.
|
||||
const inverseCells = styled[start]!.inverseCells;
|
||||
const cursorBlank = prompt.glyph === "❯" && rows[0] === " " && inverseCells[prefix] && !inverseCells[prefix - 1];
|
||||
const empty = atStart && rows.length === 1 && (rows[0] === "" || cursorBlank);
|
||||
const empty = atStart && rows.length === 1 && rows[0] === "";
|
||||
const last = visible.length - 1;
|
||||
const cursorAtEnd = y === start + last && x === prefix + stringWidth(visible[last] ?? "");
|
||||
const label = visible.length === 1 && COMPOSER_COLLAPSED_PASTE_PATTERN.test(visible[0]!) ? visible[0]! : null;
|
||||
@@ -1951,8 +1970,9 @@ export interface SendOpts {
|
||||
* decision), with how the seat's level would treat each input now. It returns the members that still go with this send:
|
||||
* their text and outbox ids. Never accepted from HTTP. */
|
||||
recheckMembers?: (holds: (input: AdmissionInput) => boolean, held: boolean) => { text: string; outboxIds: string[] };
|
||||
/** Internal observation of a completed input effect; never a permission callback. */
|
||||
onInputEffect?: (phase: "pasted" | "submitted") => void;
|
||||
/** Internal observation of a completed input effect; never a permission callback. A paste reports the exact text it
|
||||
* wrote, for delivery evidence. */
|
||||
onInputEffect?: (phase: "pasted" | "submitted", written?: { text: string }) => void;
|
||||
/** Internal managed lifecycle prerequisite; never accepted from HTTP send options. */
|
||||
beforeWrite?: () => void;
|
||||
/** Explicit internal composer-preservation request. Managed compaction instead uses ordinary draft-delivery defaults.
|
||||
@@ -2629,6 +2649,7 @@ export class SessionTransport {
|
||||
* agent input; a confirmed render does not prove agent consumption. */
|
||||
/** The managed compaction state OpenRig has recorded for a seat (the compaction enforcer's), for the sender's notice. */
|
||||
private compactionObserver?: (sessionName: string) => { stage: string; detail: string } | null;
|
||||
private evidenceFailureLogged = false;
|
||||
setCompactionObserver(observer: (sessionName: string) => { stage: string; detail: string } | null): void {
|
||||
this.compactionObserver = observer;
|
||||
}
|
||||
@@ -2658,6 +2679,61 @@ export class SessionTransport {
|
||||
}
|
||||
|
||||
private async sendWithoutCompactionNotice(sessionName: string, text: string, opts?: SendOpts): Promise<SendResult> {
|
||||
// The seat and occupant this send is for, frozen before it runs: a later handover doesn't relabel its evidence.
|
||||
const seat = this.evidenceSeat(sessionName);
|
||||
// The text the paste actually wrote (a terminal seat's exact text, a combined wake), not the text asked for.
|
||||
let wire: { text: string; at: string } | null = null;
|
||||
const observed: SendOpts | undefined = seat ? { ...opts, onInputEffect: (phase, written) => {
|
||||
if (phase === "pasted" && written) wire = { text: written.text, at: this.now().toISOString() };
|
||||
opts?.onInputEffect?.(phase, written);
|
||||
} } : opts;
|
||||
const result = await this.sendMarkingUrgent(sessionName, text, observed);
|
||||
this.recordDeliveryEvidence(sessionName, seat, opts, result, wire);
|
||||
return result;
|
||||
}
|
||||
|
||||
/** The seat a send is for, when evidence is recorded at all: its kind, pane and current occupant generation. */
|
||||
private evidenceSeat(sessionName: string): { kind: "tmux" | "external_cli" | "unknown"; pane: string | null; occupant: string | null } | null {
|
||||
if (!this.eventBus) return null;
|
||||
try {
|
||||
const meta = this.getSessionMeta(sessionName);
|
||||
// The kind is the node's binding's; with no session row or no binding there's no kind to name.
|
||||
const binding = meta.nodeId ? this.sessionRegistry.getBindingForNode(meta.nodeId) : null;
|
||||
return { kind: binding ? resolveSeatKind(binding.attachmentType).kind : "unknown", pane: meta.pane, occupant: meta.occupant };
|
||||
} catch {
|
||||
return { kind: "unknown", pane: null, occupant: null };
|
||||
}
|
||||
}
|
||||
|
||||
/** One additive record of what this send's result says, beside it. Best effort: a record that
|
||||
* can't be written never changes the result, holds the send or resends anything. A fact the result doesn't give is
|
||||
* unknown. Only an outbox row that exists counts as admitted. */
|
||||
private recordDeliveryEvidence(sessionName: string, seat: ReturnType<SessionTransport["evidenceSeat"]>, opts: SendOpts | undefined, result: SendResult,
|
||||
wire: { text: string; at: string } | null): void {
|
||||
if (!this.eventBus || !seat) return;
|
||||
try {
|
||||
const outboxIds = opts?.committedOutboxIds ?? result.outboxIds ?? [];
|
||||
const outbox = new OutboxHandler(this.db);
|
||||
// An outbox read that fails leaves admission unknown, never none or a partial list.
|
||||
let admissionUnknown = false;
|
||||
const admitted = outboxIds.filter((id) => { try { return Boolean(outbox.getById(id)); } catch { admissionUnknown = true; return false; } });
|
||||
const attemptId = randomUUID();
|
||||
this.eventBus.emit({
|
||||
type: "transport.delivery_evidence", sessionName: result.sessionName ?? sessionName, seatKind: seat.kind,
|
||||
pane: seat.pane, occupantGeneration: seat.occupant, messageId: opts?.deliveryId ?? outboxIds[0] ?? attemptId, attemptId,
|
||||
admitted: admissionUnknown ? "unknown" : admitted.length > 0 ? { outboxIds: admitted } : "none", written: writtenFact(result),
|
||||
terminalObserved: { outcome: result.outcome ?? null, verified: result.verified ?? null }, inputBoundary: "unknown", source: "send_result",
|
||||
...wireFingerprint(wire),
|
||||
});
|
||||
} catch (err) {
|
||||
if (!this.evidenceFailureLogged) {
|
||||
this.evidenceFailureLogged = true;
|
||||
console.warn(`[transport] delivery evidence wasn't recorded; sends are unaffected: ${(err as Error).message}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async sendMarkingUrgent(sessionName: string, text: string, opts?: SendOpts): Promise<SendResult> {
|
||||
if (!opts?.urgent) return this.sendClassified(sessionName, text, opts);
|
||||
// Shown on delivery, and kept with a held message: who marked it urgent.
|
||||
console.info(`[transport] urgent send to ${sessionName}, marked by ${opts.urgent.by}`);
|
||||
@@ -3652,7 +3728,7 @@ export class SessionTransport {
|
||||
...(waitMode || policyRefusal ? { sent: false, ...waitEvidence } : {}),
|
||||
});
|
||||
}
|
||||
try { opts?.onInputEffect?.("pasted"); } catch { /* Observation cannot change a completed paste. */ }
|
||||
try { opts?.onInputEffect?.("pasted", { text }); } catch { /* Observation cannot change a completed paste. */ }
|
||||
|
||||
// From the successful paste to Enter the text may sit in the seat's input. Any throw in this span is
|
||||
// tagged inputWritten, so no caller reports it as unsent; a step added here later stays covered.
|
||||
|
||||
@@ -123,6 +123,11 @@ export interface Session {
|
||||
|
||||
// -- Event types --
|
||||
|
||||
/** What a delivery-evidence record says was admitted: the outbox rows that exist, none, or unknown when they couldn't be read. */
|
||||
export type DeliveryEvidenceAdmitted = { outboxIds: string[] } | "none" | "unknown";
|
||||
/** A send's own screen check as it reported it, or nulls when none ran. */
|
||||
export type DeliveryEvidenceObserved = { outcome: string | null; verified: boolean | null };
|
||||
|
||||
export type RigEvent =
|
||||
| { type: "scope.sources_changed"; scope: string }
|
||||
| { type: "proof.judged" | "proof.sources_changed"; scope: string; revision: string }
|
||||
@@ -182,6 +187,22 @@ export type RigEvent =
|
||||
// is preserved in the log; this event records the audit trail of the upgrade.
|
||||
| { type: "restore.outcome_reconciled"; rigId: string; nodeId: string; attemptId: number; from: "failed" | "attention_required"; to: "operator_recovered"; evidence: { tmux: boolean; fgProcess: "claude" | "codex" | string; resumeTokenUsed: boolean; paneState: "usable" } | { source: string; reason?: string; kind?: string; state?: string; runtimeCwdVerified?: boolean } }
|
||||
| { type: "agent.activity"; rigId: string; nodeId: string; sessionName: string; runtime: string | null; activity: AgentActivity }
|
||||
/** What one send's own result says was admitted, written and seen on screen, for the seat and
|
||||
* occupant it was sent to. Additive: it sits beside the send's result and never rewrites it. `written` is what the
|
||||
* result shows reached the input; `terminalObserved` is today's screen check, verbatim; `inputBoundary` is unknown
|
||||
* until a harness hook is correlated. Never acceptance, a claim or completion. It names no node, so a node's own event
|
||||
* reads never see it: the session and occupant generation name the seat. */
|
||||
| { type: "transport.delivery_evidence"; sessionName: string; seatKind: "tmux" | "external_cli" | "unknown";
|
||||
pane: string | null; occupantGeneration: string | null; messageId: string; attemptId: string;
|
||||
admitted: DeliveryEvidenceAdmitted;
|
||||
written: "submitted" | "submit_only" | "text_only" | "text_submit_unknown" | "uncertain" | "none" | "unknown";
|
||||
terminalObserved: DeliveryEvidenceObserved; inputBoundary: "unknown"; source: "send_result";
|
||||
wireSha256: string | null; wireBytes: number | null; writtenAt: string | null }
|
||||
/** A harness's own prompt hook matched to the one message OpenRig wrote (same session and occupant generation, an equal
|
||||
* text fingerprint, a single candidate): that message reached the input boundary. Never a claim, completion or
|
||||
* acceptance. It names no node. */
|
||||
| { type: "transport.input_boundary"; sessionName: string; occupantGeneration: string; messageId: string; attemptId: string;
|
||||
observed: "UserPromptSubmit"; runtime: string | null; nativePromptId: string | null; nativeTurnId: string | null; observedAt: string }
|
||||
// OPR.0.4.1.10 — audit record for a --dangerously-interact send that drove a target's interactive
|
||||
// prompt / permission block. overrideReason = caller's --reason; detectedReason/evidenceSource =
|
||||
// what the classifier saw (kept distinct so permission_prompt vs selection_prompt vs unknown stays visible).
|
||||
|
||||
@@ -7,6 +7,7 @@ import type { AgentActivityStore } from "../domain/agent-activity-store.js";
|
||||
import { hookEventState, isRunnerExit } from "../domain/agent-activity-store.js";
|
||||
import { WorkingLivenessGate, isWorkingLivenessBeat, workingLivenessSeconds, type WorkingLivenessIgnored } from "../domain/working-liveness.js";
|
||||
import type { EventBus } from "../domain/event-bus.js";
|
||||
import { inputBoundaryCorrelatorFor } from "../domain/input-boundary-correlator.js";
|
||||
import type { StreamStore } from "../domain/stream-store.js";
|
||||
import { verifyStartupProof } from "../domain/startup-proof.js";
|
||||
import { ActivityGrantStore, grantGeneration, type ActivityGrant } from "../domain/activity-grants.js";
|
||||
@@ -733,6 +734,7 @@ function recordHookActivity(c: Context, store: AgentActivityStore, body: Record<
|
||||
const status = result.code === "missing_session_identity" ? 400 : 404;
|
||||
return { refused: c.json({ ok: false, code: result.code, error: result.error }, status) };
|
||||
}
|
||||
noteInputBoundary(c, body, (result.event as { sessionName?: string } | undefined)?.sessionName ?? stringOrNull(body.sessionName));
|
||||
|
||||
// S19 A4 — feed the ONE oracle through the adapter seam: the recorded (store-
|
||||
// normalized) event becomes ladder evidence on the lifecycle-hooks rung. The store
|
||||
@@ -1073,6 +1075,24 @@ activityRoutes.get("/parked", (c) => {
|
||||
});
|
||||
});
|
||||
|
||||
/** A prompt hook's input-boundary facts (a fingerprint of the prompt the harness received), handed to the correlator that
|
||||
* matches them to the message OpenRig wrote. Malformed or missing facts are dropped, never refused: the boundary stays
|
||||
* unknown. */
|
||||
function noteInputBoundary(c: Context, body: Record<string, unknown>, sessionName: string | null): void {
|
||||
if (body.hookEvent !== "UserPromptSubmit") return;
|
||||
const eventBus = c.get("eventBus" as never) as EventBus | undefined;
|
||||
const generation = stringOrNull(body.generation);
|
||||
const occurredAt = stringOrNull(body.occurredAt);
|
||||
const sha256 = typeof body.promptSha256 === "string" && /^[0-9a-f]{64}$/.test(body.promptSha256) ? body.promptSha256 : null;
|
||||
const bytes = typeof body.promptBytes === "number" && Number.isSafeInteger(body.promptBytes) && body.promptBytes >= 0 ? body.promptBytes : null;
|
||||
if (!eventBus || !sessionName || !generation || !occurredAt || !sha256 || bytes === null) return;
|
||||
const nativeId = (value: unknown) => typeof value === "string" && /^[A-Za-z0-9_-]{1,128}$/.test(value) ? value : null;
|
||||
try {
|
||||
inputBoundaryCorrelatorFor(eventBus).noteHook({ sessionName, generation, sha256, bytes, occurredAt, runtime: stringOrNull(body.runtime),
|
||||
nativePromptId: nativeId(body.nativePromptId), nativeTurnId: nativeId(body.turnId) });
|
||||
} catch { /* Best effort: the hook is recorded either way. */ }
|
||||
}
|
||||
|
||||
function stringOrNull(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
@@ -8,6 +8,10 @@ export const eventsRoute = new Hono();
|
||||
/** A replay is read and written this many events at a time, never built whole. */
|
||||
const REPLAY_PAGE = 500;
|
||||
|
||||
/** Event types this stream doesn't carry, live or replayed: one-per-send diagnostics the log keeps for their own readers,
|
||||
* which would otherwise fill a viewer's newest-N replay and its cache. */
|
||||
const OMITTED_TYPES: readonly string[] = ["transport.delivery_evidence", "transport.input_boundary"];
|
||||
|
||||
function getEventBus(c: { get: (key: string) => unknown }): EventBus {
|
||||
return c.get("eventBus" as never) as EventBus;
|
||||
}
|
||||
@@ -36,6 +40,7 @@ eventsRoute.get("/", (c) => {
|
||||
// A transient event (view.changed) is never stored, so it is never replayed and its seq is its cause's: this
|
||||
// stream carries stored events only, live and replayed alike.
|
||||
if (event.transient) return;
|
||||
if (OMITTED_TYPES.includes(event.type)) return;
|
||||
// Rig-scoped: filter by rigId. Global: accept all events.
|
||||
if (rigId) {
|
||||
const eventRigId = "rigId" in event ? (event as { rigId: string }).rigId : null;
|
||||
@@ -59,15 +64,15 @@ eventsRoute.get("/", (c) => {
|
||||
// Last-Event-ID, `?replay=<n>` replays the newest n (0: from now on), the whole log when it holds no more.
|
||||
const lastSeq = lastEventId !== null ? (isNaN(lastEventId) ? 0 : lastEventId)
|
||||
: replayCount === null ? 0
|
||||
: eventBus.seqBeforeNewest(replayCount, rigId);
|
||||
: eventBus.seqBeforeNewest(replayCount, rigId, OMITTED_TYPES);
|
||||
maxReplayedSeq = lastSeq;
|
||||
// Where it ends, read once subscribed too: an event stored after it is buffered, so events arriving while the
|
||||
// pages are written can neither keep the replay going nor be sent twice.
|
||||
const upTo = eventBus.maxSeq();
|
||||
|
||||
// 2. Replay missed events from DB, a page at a time
|
||||
for (let page = eventBus.replayPage(lastSeq, upTo, REPLAY_PAGE, rigId); page.length > 0 && !stream.aborted;
|
||||
page = eventBus.replayPage(maxReplayedSeq, upTo, REPLAY_PAGE, rigId)) {
|
||||
for (let page = eventBus.replayPage(lastSeq, upTo, REPLAY_PAGE, rigId, OMITTED_TYPES); page.length > 0 && !stream.aborted;
|
||||
page = eventBus.replayPage(maxReplayedSeq, upTo, REPLAY_PAGE, rigId, OMITTED_TYPES)) {
|
||||
for (const event of page) {
|
||||
await stream.writeSSE({ id: String(event.seq), data: JSON.stringify(event) });
|
||||
if (event.seq > maxReplayedSeq) maxReplayedSeq = event.seq;
|
||||
|
||||
@@ -39,6 +39,7 @@ import { resolveExternal } from "./domain/gateway/external-admission.js";
|
||||
import { makeTopologyDestinationCheck } from "./domain/queue-destination-check.js";
|
||||
import { loadHumanRegistry } from "./domain/gateway/human-registry.js";
|
||||
import { EventBus } from "./domain/event-bus.js";
|
||||
import { inputBoundaryCorrelatorFor } from "./domain/input-boundary-correlator.js";
|
||||
import { NodeLauncher } from "./domain/node-launcher.js";
|
||||
import { TmuxOptionDefaultsApplier } from "./domain/tmux-option-defaults.js";
|
||||
import { TmuxAdapter } from "./adapters/tmux.js";
|
||||
@@ -386,6 +387,8 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
|
||||
const rigRepo = new RigRepository(db);
|
||||
const sessionRegistry = new SessionRegistry(db);
|
||||
const eventBus = new EventBus(db);
|
||||
// Subscribed from the start, so every written message can be matched to its harness's own prompt hook.
|
||||
inputBoundaryCorrelatorFor(eventBus);
|
||||
const streamStore = new StreamStore(db, eventBus);
|
||||
// Explicit source opt-in only. No sink I/O until a separate drain request.
|
||||
const shadow = configureShadowCapture(process.env.OPENRIG_SHADOW_CAPTURE);
|
||||
|
||||
@@ -6,6 +6,7 @@ import { createRequire } from "node:module";
|
||||
import fs from "node:fs";
|
||||
import os from "node:os";
|
||||
import nodePath from "node:path";
|
||||
import { createHash } from "node:crypto";
|
||||
|
||||
const require = createRequire(import.meta.url);
|
||||
const relay = require("../assets/plugins/openrig-core/hooks/scripts/activity-relay.cjs") as {
|
||||
@@ -94,3 +95,24 @@ describe("activity-relay occupant generation carry (W2a producer)", () => {
|
||||
expect(payload).toHaveProperty("generation", null);
|
||||
});
|
||||
});
|
||||
|
||||
describe("activity-relay prompt fingerprint at UserPromptSubmit", () => {
|
||||
const identity = { OPENRIG_SESSION_NAME: "dev-impl@r", OPENRIG_RUNTIME: "claude-code", OPENRIG_OCCUPANT_GENERATION: "gen-1" };
|
||||
const at = () => new Date("2026-11-11T03:00:00.000Z");
|
||||
|
||||
it("carries the prompt's fingerprint, byte length and Claude's prompt id, never the prompt", () => {
|
||||
const payload = relay.buildOpenRigPayload({ hook_event_name: "UserPromptSubmit", prompt: "héllo, world", prompt_id: "p-1", session_id: "s-1" }, identity, at)!;
|
||||
expect(payload).toMatchObject({
|
||||
hookEvent: "UserPromptSubmit", generation: "gen-1", nativePromptId: "p-1", promptBytes: Buffer.byteLength("héllo, world"),
|
||||
promptSha256: createHash("sha256").update("héllo, world").digest("hex"),
|
||||
});
|
||||
expect(JSON.stringify(payload)).not.toContain("world");
|
||||
});
|
||||
|
||||
it("adds nothing for another event, a prompt that isn't a string, or a malformed prompt id", () => {
|
||||
expect(relay.buildOpenRigPayload({ hook_event_name: "Stop", prompt: "x" }, identity, at)).not.toHaveProperty("promptSha256");
|
||||
expect(relay.buildOpenRigPayload({ hook_event_name: "UserPromptSubmit", prompt: 7 }, identity, at)).not.toHaveProperty("promptSha256");
|
||||
expect(relay.buildOpenRigPayload({ hook_event_name: "UserPromptSubmit", prompt: "x", prompt_id: "not ok/../" }, identity, at))
|
||||
.not.toHaveProperty("nativePromptId");
|
||||
});
|
||||
});
|
||||
@@ -162,15 +162,6 @@ describe("cursor-bound composer input", () => {
|
||||
expect(inspectComposerInput(claude("\x1b[2;38;5;245mAsk Codex to do anything\x1b[0m", { x: 2, y: 2 })).state).toBe("empty");
|
||||
});
|
||||
|
||||
it("recognizes only a single inverse-video Claude cursor blank at the prompt start", () => {
|
||||
const cursor = { x: 2, y: 2 };
|
||||
expect(inspectComposerInput(claude("\x1b[7m \x1b[0m", cursor))).toMatchObject({ state: "empty", rows: [""], cursorAtEnd: true });
|
||||
for (const body of [" ", "\x1b[38;5;7m \x1b[0m", "\x1b[7m\x1b[27m ", "\x1b[7m \x1b[0mdraft"]) {
|
||||
expect(inspectComposerInput(claude(body, cursor)).state).toBe("text");
|
||||
}
|
||||
expect(inspectComposerInput(claude("\x1b[7m \x1b[0m", { x: 3, y: 2 }))).toMatchObject({ state: "text", rows: [" "] });
|
||||
});
|
||||
|
||||
it("requires the complete owned text and preserves significant spaces", () => {
|
||||
const input = inspectComposerInput(claude("Run two commands\nthen report."));
|
||||
expect(composerContainsOwnedText(input, "Run two commands\nthen report.")).toBe(true);
|
||||
|
||||
@@ -10,7 +10,7 @@ import { TmuxAdapter } from "../src/adapters/tmux.js";
|
||||
import { shellQuote } from "../src/adapters/shell-quote.js";
|
||||
import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
|
||||
import { migrate } from "../src/db/migrate.js";
|
||||
import { annotateInputBox, inspectComposerInput } from "../src/domain/session-transport.js";
|
||||
import { inspectComposerInput } from "../src/domain/session-transport.js";
|
||||
import { RigRepository } from "../src/domain/rig-repository.js";
|
||||
import { SeatDeliveryGuard, resolveGuardTarget } from "../src/domain/seat-delivery-guard.js";
|
||||
import { SessionRegistry } from "../src/domain/session-registry.js";
|
||||
@@ -133,7 +133,7 @@ describe.skipIf(process.platform === "win32")("native draft-aware terminal deliv
|
||||
}
|
||||
}, 30_000);
|
||||
|
||||
it.each(["update-banner", "cursor-block"])("submits a multiline paste with %s and keeps a real draft through tmux", async variant => {
|
||||
it("submits a multiline paste under a combined update footer through real tmux", async () => {
|
||||
const scratch = mkdtempSync(join(tmpdir(), "draft-banner-"));
|
||||
const socket = join(scratch, "s");
|
||||
const env = { ...process.env }; delete env.TMUX; delete env.TMUX_TMPDIR;
|
||||
@@ -143,9 +143,8 @@ describe.skipIf(process.platform === "win32")("native draft-aware terminal deliv
|
||||
const db = new Database(":memory:");
|
||||
let transport: SessionTransport | undefined;
|
||||
try {
|
||||
const command = [process.execPath, composer, "claude", stateFile, variant].map(shellQuote).join(" ");
|
||||
const [width, height] = variant === "cursor-block" ? ["68", "31"] : ["120", "40"];
|
||||
const pane = (await tmux(["-f", "/dev/null", "new-session", "-d", "-P", "-F", "#{pane_id}", "-x", width, "-y", height, "-s", "worker@test", command])).trim();
|
||||
const command = [process.execPath, composer, "claude", stateFile, "update-banner"].map(shellQuote).join(" ");
|
||||
const pane = (await tmux(["-f", "/dev/null", "new-session", "-d", "-P", "-F", "#{pane_id}", "-x", "120", "-y", "40", "-s", "worker@test", command])).trim();
|
||||
await vi.waitFor(() => expect(state().input).toBe(""), { timeout: 5000 });
|
||||
migrate(db, ALL_MIGRATIONS);
|
||||
db.exec(`INSERT INTO rigs(id,name) VALUES ('rig','test');
|
||||
@@ -158,14 +157,7 @@ describe.skipIf(process.platform === "win32")("native draft-aware terminal deliv
|
||||
const guard = new SeatDeliveryGuard(db, name => resolveGuardTarget(db, name)); adapter.deliveryGuard = guard;
|
||||
transport = new SessionTransport({ db, tmuxAdapter: adapter, rigRepo: new RigRepository(db), sessionRegistry: new SessionRegistry(db) });
|
||||
await guard.set("a", { mode: "draft-aware" }, "human@test", "protect fixture input");
|
||||
if (variant === "cursor-block") {
|
||||
await vi.waitFor(async () => {
|
||||
const snapshot = await adapter.captureComposerSnapshot(pane);
|
||||
expect(snapshot?.cursor).toEqual({ x: 2, y: 26, width: 68, height: 31 });
|
||||
expect(snapshot!.screen.split("\n")[26]).toContain("❯\u00a0\x1b[7m ");
|
||||
expect(annotateInputBox(snapshot!.screen, snapshot!.cursor, "claude-code")).toMatchObject({ status: "empty", confidence: "high" });
|
||||
});
|
||||
} else await vi.waitFor(async () => expect(inspectComposerInput(await adapter.captureComposerSnapshot(pane), "claude-code").state).toBe("empty"));
|
||||
await vi.waitFor(async () => expect(inspectComposerInput(await adapter.captureComposerSnapshot(pane), "claude-code").state).toBe("empty"));
|
||||
const text = "first line\nsecond line\nlast line";
|
||||
const result = await transport.send("worker@test", text, { deliveryId: "banner", actorSession: "sender@test", verify: false });
|
||||
expect(result, JSON.stringify(result)).toMatchObject({ ok: true, sent: true });
|
||||
@@ -173,7 +165,7 @@ describe.skipIf(process.platform === "win32")("native draft-aware terminal deliv
|
||||
await tmux(["send-keys", "-t", pane, "-l", "person's draft"]);
|
||||
await vi.waitFor(() => expect(state().input).toBe("person's draft"));
|
||||
expect(await transport.send("worker@test", "another message", { deliveryId: "draft", verify: false }))
|
||||
.toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
|
||||
.toMatchObject({ outcome: "retained", sent: false });
|
||||
expect(state()).toEqual({ input: "person's draft", submissions: [text] });
|
||||
} finally {
|
||||
await transport?.guardedDelivery?.stop();
|
||||
|
||||
@@ -262,6 +262,34 @@ describe("SSE events route", () => {
|
||||
expect(ids(await readSSEEvents(res, 3))).toEqual(seqs.slice(1));
|
||||
});
|
||||
|
||||
const evidence = (i: number) => eventBus.emit({ type: "transport.delivery_evidence", sessionName: "dev-impl@my-rig", seatKind: "tmux", pane: null,
|
||||
occupantGeneration: null, messageId: `m${i}`, attemptId: `a${i}`, admitted: "none", written: "submitted",
|
||||
terminalObserved: { outcome: null, verified: null }, inputBoundary: "unknown", source: "send_result", wireSha256: null, wireBytes: null, writtenAt: null }).seq;
|
||||
|
||||
it("a fresh replay=<n> skips per-send delivery evidence, so sends after the newest real events never empty a new feed", async () => {
|
||||
const rig = rigRepo.createRig("r01");
|
||||
const seqs = emitMany(rig.id, 2);
|
||||
const stored = Array.from({ length: 5 }, (_, i) => evidence(i));
|
||||
// The log keeps them for their own readers.
|
||||
expect(db.prepare("SELECT COUNT(*) AS n FROM events WHERE type = 'transport.delivery_evidence'").get()).toEqual({ n: stored.length });
|
||||
expect(ids(await readSSEEvents(await app.request("/api/events?replay=2"), 3))).toEqual(seqs);
|
||||
});
|
||||
|
||||
it("live delivery evidence and input-boundary records aren't written to the stream, so a viewer's cache never holds them, and a reconnect's replay skips them", async () => {
|
||||
const rig = rigRepo.createRig("r01");
|
||||
const [first] = emitMany(rig.id, 1);
|
||||
const live = await app.request("/api/events?replay=0");
|
||||
setTimeout(() => {
|
||||
evidence(0);
|
||||
eventBus.emit({ type: "transport.input_boundary", sessionName: "dev-impl@my-rig", occupantGeneration: "g", messageId: "m0", attemptId: "a0",
|
||||
observed: "UserPromptSubmit", runtime: "claude-code", nativePromptId: null, nativeTurnId: null, observedAt: new Date().toISOString() });
|
||||
eventBus.emit({ type: "rig.created", rigId: rig.id });
|
||||
}, 10);
|
||||
expect((await readSSEEvents(live, 2)).map((e) => JSON.parse(e.data).type)).toEqual(["rig.created"]);
|
||||
const again = await app.request("/api/events?replay=0", { headers: { "Last-Event-ID": String(first) } });
|
||||
expect((await readSSEEvents(again, 3)).map((e) => JSON.parse(e.data).type)).toEqual(["rig.created"]);
|
||||
});
|
||||
|
||||
it("the UI's first connection, replay=100 over 1,000 stored events, gets the newest 100 in order, then live", async () => {
|
||||
const rig = rigRepo.createRig("r01");
|
||||
const seqs = emitMany(rig.id, 1000);
|
||||
|
||||
+3
-7
@@ -5,23 +5,19 @@ import { StringDecoder } from "node:string_decoder";
|
||||
import stringWidth from "string-width";
|
||||
|
||||
const [runtime, stateFile, footerVariant] = process.argv.slice(2);
|
||||
const cursorBlock = footerVariant === "cursor-block";
|
||||
const rule = cursorBlock ? `\x1b[38;5;244m${"─".repeat(68)}\x1b[39m` : "────────────────────────────────────────";
|
||||
let input = "", pending = "", pasting = false, cursor = 0, ghost = false;
|
||||
const submissions = [];
|
||||
const decoder = new StringDecoder("utf8");
|
||||
|
||||
function render() {
|
||||
const rows = ["Editable delivery fixture", ...submissions.slice(-3).flatMap(text => text.split("\n"))];
|
||||
if (cursorBlock) while (rows.length < 25) rows.push("");
|
||||
if (runtime === "claude") rows.push(rule);
|
||||
if (runtime === "claude") rows.push("────────────────────────────────────────");
|
||||
const start = rows.length;
|
||||
const displayed = ghost && cursor === input.length ? `${input}\x1b[2m suggested follow-up\x1b[22m`
|
||||
: input || (runtime === "codex" ? "\x1b[2mAsk Codex to do anything\x1b[22m" : cursorBlock ? "\x1b[7m \x1b[0m" : "");
|
||||
: input || (runtime === "codex" ? "\x1b[2mAsk Codex to do anything\x1b[22m" : "");
|
||||
const body = displayed.split("\n");
|
||||
rows.push(`${runtime === "claude" ? "❯\u00a0" : "› "}${body[0]}`, ...body.slice(1).map(line => ` ${line}`));
|
||||
rows.push(...(runtime === "claude" ? [rule, cursorBlock
|
||||
? " \x1b[38;5;211m⏵⏵ bypass permissions on\x1b[38;5;246m (shift+tab to cycle)\x1b[0m" : "? for shortcuts"] : ["", "fixture · 90% context left"]));
|
||||
rows.push(...(runtime === "claude" ? ["────────────────────────────────────────", "? for shortcuts"] : ["", "fixture · 90% context left"]));
|
||||
if (runtime === "claude" && ghost) {
|
||||
rows[rows.length - 1] = "⏵⏵ accept edits on (shift+tab to cycle) · ← for agents";
|
||||
rows.push("✘ Auto-update failed: no write permission to npm prefix · Run claude doctor");
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
// A harness's prompt hook is matched to the one message OpenRig wrote only when the match is unambiguous; anything else
|
||||
// leaves the input boundary unknown.
|
||||
import { describe, it, expect, beforeEach } from "vitest";
|
||||
import type Database from "better-sqlite3";
|
||||
import { createFullTestDb } from "./helpers/test-app.js";
|
||||
import { EventBus } from "../src/domain/event-bus.js";
|
||||
import { InputBoundaryCorrelator, type BoundaryHook } from "../src/domain/input-boundary-correlator.js";
|
||||
|
||||
const SHA = "a".repeat(64);
|
||||
const OTHER = "b".repeat(64);
|
||||
|
||||
describe("InputBoundaryCorrelator", () => {
|
||||
let db: Database.Database;
|
||||
let bus: EventBus;
|
||||
let clock: number;
|
||||
let correlator: InputBoundaryCorrelator;
|
||||
|
||||
beforeEach(() => {
|
||||
db = createFullTestDb();
|
||||
bus = new EventBus(db);
|
||||
clock = Date.parse("2026-11-11T03:00:00.000Z");
|
||||
correlator = new InputBoundaryCorrelator(bus, () => clock);
|
||||
bus.subscribe((event) => correlator.noteRecord(event));
|
||||
});
|
||||
|
||||
const write = (over: Partial<{ messageId: string; generation: string | null; sha: string; bytes: number; writtenAt: string; written: string }> = {}) =>
|
||||
bus.emit({
|
||||
type: "transport.delivery_evidence", sessionName: "dev-impl@r", seatKind: "tmux", pane: "%1",
|
||||
occupantGeneration: over.generation === undefined ? "gen-1" : over.generation, messageId: over.messageId ?? "m1", attemptId: `a-${over.messageId ?? "m1"}`,
|
||||
admitted: "none", written: (over.written ?? "submitted") as "submitted", terminalObserved: { outcome: null, verified: null },
|
||||
inputBoundary: "unknown", source: "send_result", wireSha256: over.sha ?? SHA, wireBytes: over.bytes ?? 5,
|
||||
writtenAt: over.writtenAt ?? "2026-11-11T03:00:00.000Z",
|
||||
});
|
||||
const hook = (over: Partial<BoundaryHook> = {}) => correlator.noteHook({
|
||||
sessionName: "dev-impl@r", generation: "gen-1", sha256: SHA, bytes: 5, occurredAt: "2026-11-11T03:00:01.000Z",
|
||||
runtime: "claude-code", nativePromptId: "p-1", nativeTurnId: null, ...over,
|
||||
});
|
||||
const observed = () => (db.prepare("SELECT node_id, payload FROM events WHERE type = 'transport.input_boundary' ORDER BY seq").all() as Array<{ node_id: string | null; payload: string }>)
|
||||
.map((row) => ({ nodeId: row.node_id, ...JSON.parse(row.payload) }));
|
||||
|
||||
it("records the boundary for the one write its hook matches, written first or second, with no node", () => {
|
||||
write({ messageId: "m1" });
|
||||
hook();
|
||||
hook({ occurredAt: "2026-11-11T03:00:03.000Z", nativePromptId: "p-2" });
|
||||
write({ messageId: "m2", writtenAt: "2026-11-11T03:00:02.000Z" });
|
||||
expect(observed()).toMatchObject([
|
||||
{ nodeId: null, messageId: "m1", attemptId: "a-m1", occupantGeneration: "gen-1", observed: "UserPromptSubmit", runtime: "claude-code", nativePromptId: "p-1" },
|
||||
{ nodeId: null, messageId: "m2", nativePromptId: "p-2" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("leaves the boundary unknown for two candidates, another generation, no generation, another text or length, or a write after the hook", () => {
|
||||
write({ messageId: "m1" });
|
||||
write({ messageId: "m2" });
|
||||
hook();
|
||||
write({ messageId: "g", sha: OTHER, generation: "gen-2" });
|
||||
hook({ sha256: OTHER });
|
||||
write({ messageId: "n", sha: "c".repeat(64), generation: null });
|
||||
hook({ sha256: "c".repeat(64) });
|
||||
write({ messageId: "t", sha: "d".repeat(64) });
|
||||
hook({ sha256: "e".repeat(64) });
|
||||
write({ messageId: "l", sha: "f".repeat(64), bytes: 6 });
|
||||
hook({ sha256: "f".repeat(64) });
|
||||
write({ messageId: "late", sha: "1".repeat(64), writtenAt: "2026-11-11T03:00:05.000Z" });
|
||||
hook({ sha256: "1".repeat(64), occurredAt: "2026-11-11T03:00:04.000Z" });
|
||||
hook({ sha256: "3".repeat(64), occurredAt: "2026-11-11T03:00:04.000Z" });
|
||||
write({ messageId: "late-after-hook", sha: "3".repeat(64), writtenAt: "2026-11-11T03:00:05.000Z" });
|
||||
expect(observed()).toEqual([]);
|
||||
});
|
||||
|
||||
it("matches a write once, never a write whose text didn't reach the input, and nothing older than the window", () => {
|
||||
write({ messageId: "m1" });
|
||||
hook();
|
||||
hook({ nativePromptId: "p-again" });
|
||||
write({ messageId: "u", sha: OTHER, written: "uncertain" });
|
||||
hook({ sha256: OTHER });
|
||||
write({ messageId: "old", sha: "2".repeat(64) });
|
||||
clock += 11 * 60_000;
|
||||
hook({ sha256: "2".repeat(64), occurredAt: "2026-11-11T03:11:00.000Z" });
|
||||
expect(observed().map((event) => event.messageId)).toEqual(["m1"]);
|
||||
});
|
||||
|
||||
it("matches nothing after a restart: a new correlator holds none of the writes before it", () => {
|
||||
write({ messageId: "m1" });
|
||||
const restarted = new InputBoundaryCorrelator(bus, () => clock);
|
||||
restarted.noteHook({ sessionName: "dev-impl@r", generation: "gen-1", sha256: SHA, bytes: 5, occurredAt: "2026-11-11T03:00:01.000Z",
|
||||
runtime: "codex", nativePromptId: null, nativeTurnId: "t-1" });
|
||||
expect(observed()).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,62 @@
|
||||
// The hook route hands a prompt hook's fingerprint to the correlator: a matching write records the input boundary, and a
|
||||
// malformed fingerprint is dropped without refusing the hook.
|
||||
import { afterEach, describe, it, expect } from "vitest";
|
||||
import { Hono } from "hono";
|
||||
import { createFullTestDb } from "./helpers/test-app.js";
|
||||
import { EventBus } from "../src/domain/event-bus.js";
|
||||
import { inputBoundaryCorrelatorFor } from "../src/domain/input-boundary-correlator.js";
|
||||
import { activityRoutes } from "../src/routes/activity.js";
|
||||
|
||||
const TOKEN = "test-token";
|
||||
const SESSION = "dev-impl@r";
|
||||
const SHA = "a".repeat(64);
|
||||
const closeDbs: Array<() => void> = [];
|
||||
afterEach(() => { for (const close of closeDbs.splice(0)) close(); });
|
||||
|
||||
function makeApp() {
|
||||
const db = createFullTestDb();
|
||||
closeDbs.push(() => db.close());
|
||||
const bus = new EventBus(db);
|
||||
inputBoundaryCorrelatorFor(bus);
|
||||
const store = { db, recordHookEvent: () => ({ ok: true as const, activity: { state: "running", reason: "prompt", evidenceSource: "hook", sampledAt: new Date().toISOString(), evidence: null, runtime: "claude-code" }, event: {} }) };
|
||||
const app = new Hono();
|
||||
app.use("*", async (c, next) => {
|
||||
c.set("agentActivityStore" as never, store as never);
|
||||
c.set("activityHookToken" as never, TOKEN as never);
|
||||
c.set("eventBus" as never, bus as never);
|
||||
await next();
|
||||
});
|
||||
app.route("/api/activity", activityRoutes);
|
||||
const post = (fields: Record<string, unknown>) => app.request("/api/activity/hooks", {
|
||||
method: "POST",
|
||||
headers: { "content-type": "application/json", authorization: `Bearer ${TOKEN}` },
|
||||
body: JSON.stringify({ runtime: "claude-code", sessionName: SESSION, hookEvent: "UserPromptSubmit", generation: "gen-1",
|
||||
occurredAt: "2026-11-11T03:00:01.000Z", ...fields }),
|
||||
});
|
||||
const written = () => bus.emit({
|
||||
type: "transport.delivery_evidence", sessionName: SESSION, seatKind: "tmux", pane: "%1", occupantGeneration: "gen-1",
|
||||
messageId: "m1", attemptId: "a1", admitted: "none", written: "submitted", terminalObserved: { outcome: null, verified: null },
|
||||
inputBoundary: "unknown", source: "send_result", wireSha256: SHA, wireBytes: 5, writtenAt: "2026-11-11T03:00:00.000Z",
|
||||
});
|
||||
const observed = () => db.prepare("SELECT payload FROM events WHERE type = 'transport.input_boundary'").all() as Array<{ payload: string }>;
|
||||
return { post, written, observed };
|
||||
}
|
||||
|
||||
describe("the hook route and the input boundary", () => {
|
||||
it("records the boundary when a prompt hook's fingerprint matches the one write", async () => {
|
||||
const { post, written, observed } = makeApp();
|
||||
written();
|
||||
const res = await post({ promptSha256: SHA, promptBytes: 5, nativePromptId: "p-1" });
|
||||
expect(res.status).toBe(200);
|
||||
expect(observed().map((row) => JSON.parse(row.payload))).toMatchObject([{ messageId: "m1", nativePromptId: "p-1", observed: "UserPromptSubmit" }]);
|
||||
});
|
||||
|
||||
it("drops a malformed fingerprint or a missing generation and still records the hook", async () => {
|
||||
const { post, written, observed } = makeApp();
|
||||
written();
|
||||
expect((await post({ promptSha256: "not-a-sha", promptBytes: 5 })).status).toBe(200);
|
||||
expect((await post({ promptSha256: SHA, promptBytes: -1 })).status).toBe(200);
|
||||
expect((await post({ promptSha256: SHA, promptBytes: 5, generation: null })).status).toBe(200);
|
||||
expect(observed()).toEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -2,6 +2,7 @@ import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
|
||||
import type Database from "better-sqlite3";
|
||||
import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { createHash } from "node:crypto";
|
||||
import { join } from "node:path";
|
||||
import { SettingsStore } from "../src/domain/user-settings/settings-store.js";
|
||||
import { createDb } from "../src/db/connection.js";
|
||||
@@ -19,7 +20,7 @@ import { agentspecRebootSchema } from "../src/db/migrations/014_agentspec_reboot
|
||||
import { externalCliAttachmentSchema } from "../src/db/migrations/019_external_cli_attachment.js";
|
||||
import { RigRepository } from "../src/domain/rig-repository.js";
|
||||
import { SessionRegistry } from "../src/domain/session-registry.js";
|
||||
import { classifyPaneActivity, classifyStyledFrame, draftDeliveryAllows, readDraftDeliverySetting, hasExpectedStagedText, inspectStartupStagedText, probeSessionActivity, promptTextIsFaint, SessionTransport, type TransportReadinessChecker } from "../src/domain/session-transport.js";
|
||||
import { classifyPaneActivity, classifyStyledFrame, draftDeliveryAllows, readDraftDeliverySetting, hasExpectedStagedText, inspectStartupStagedText, probeSessionActivity, promptTextIsFaint, SessionTransport, writtenFact, type TransportReadinessChecker } from "../src/domain/session-transport.js";
|
||||
import { SeatDeliveryGuard, resolveGuardTarget, type GuardTarget } from "../src/domain/seat-delivery-guard.js";
|
||||
import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
|
||||
import { outboxEntriesSchema } from "../src/db/migrations/027_outbox_entries.js";
|
||||
@@ -27,6 +28,7 @@ import { seatDeliveryGuardSchema } from "../src/db/migrations/087_seat_delivery_
|
||||
import { AgentActivityStore } from "../src/domain/agent-activity-store.js";
|
||||
import { EventBus } from "../src/domain/event-bus.js";
|
||||
import { QueueRepository } from "../src/domain/queue-repository.js";
|
||||
import { OutboxHandler } from "../src/domain/outbox-handler.js";
|
||||
import { getNodeInventory } from "../src/domain/node-inventory.js";
|
||||
import type { TmuxAdapter, TmuxResult } from "../src/adapters/tmux.js";
|
||||
import { createFullTestDb } from "./helpers/test-app.js";
|
||||
@@ -2762,6 +2764,185 @@ describe("SessionTransport", () => {
|
||||
expect(captured.content).toContain("line1");
|
||||
});
|
||||
|
||||
describe("delivery evidence", () => {
|
||||
const evidence = () => (db.prepare("SELECT payload FROM events WHERE type = 'transport.delivery_evidence' ORDER BY seq").all() as Array<{ payload: string }>)
|
||||
.map((row) => JSON.parse(row.payload));
|
||||
const typing = () => mockTmux({ capturePaneContent: async () => "hello" });
|
||||
|
||||
it("records one fact per send beside an unchanged result: the seat, its occupant, what was written and the screen check", async () => {
|
||||
seedCanonicalRig();
|
||||
const plain = await createTransport(typing()).send("dev-impl@my-rig", "hello");
|
||||
const withEvidence = await createTransport(typing(), { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello");
|
||||
expect(withEvidence).toEqual(plain);
|
||||
const [fact, ...more] = evidence();
|
||||
expect(more).toEqual([]);
|
||||
// Stored with no node, so a node's own "last N events" reads never see a per-send record.
|
||||
expect(db.prepare("SELECT node_id FROM events WHERE type = 'transport.delivery_evidence'").all()).toEqual([{ node_id: null }]);
|
||||
expect(fact).toMatchObject({ sessionName: "dev-impl@my-rig", seatKind: "tmux", admitted: "none", inputBoundary: "unknown",
|
||||
source: "send_result", terminalObserved: { outcome: withEvidence.outcome ?? null, verified: withEvidence.verified ?? null } });
|
||||
// An ordinary send issues the text and Enter and checks no screen: submitted, with no screen check recorded.
|
||||
expect(withEvidence).toMatchObject({ ok: true });
|
||||
expect(withEvidence.outcome).toBeUndefined();
|
||||
expect(fact).toMatchObject({ written: "submitted", terminalObserved: { outcome: null, verified: null } });
|
||||
expect(fact.messageId).toBe(fact.attemptId);
|
||||
});
|
||||
|
||||
it("uses the caller's delivery id as the message, and gives each attempt its own id", async () => {
|
||||
seedCanonicalRig();
|
||||
const transport = createTransport(typing(), { eventBus: new EventBus(db) });
|
||||
await transport.send("dev-impl@my-rig", "hello", { deliveryId: "msg-1" });
|
||||
await transport.send("dev-impl@my-rig", "hello", { deliveryId: "msg-1" });
|
||||
const facts = evidence();
|
||||
expect(facts.map((fact) => fact.messageId)).toEqual(["msg-1", "msg-1"]);
|
||||
expect(new Set(facts.map((fact) => fact.attemptId)).size).toBe(2);
|
||||
});
|
||||
|
||||
it("reads what was written from each result shape, and nothing the result doesn't show", () => {
|
||||
const s = "dev-impl@my-rig";
|
||||
expect([
|
||||
{ ok: true, sessionName: s },
|
||||
{ ok: true, sessionName: s, verified: true, outcome: "delivered" },
|
||||
{ ok: true, sessionName: s, verified: false, outcome: "rendered-unconfirmed" },
|
||||
{ ok: true, sessionName: s, outcome: "rendered-unconfirmed", submitOnly: true },
|
||||
{ ok: true, sessionName: s, sent: true, verified: false, outcome: "rendered-unconfirmed", promptInteraction: "unverified" },
|
||||
{ ok: true, sessionName: s, outcome: "retained", sent: false, verified: false },
|
||||
{ ok: false, sessionName: s, outcome: "failed", inputPhase: "attempted" },
|
||||
{ ok: false, sessionName: s, outcome: "failed", inputPhase: "written", reason: "submit_failed" },
|
||||
{ ok: false, sessionName: s, sent: true, outcome: "failed", inputPhase: "written", reason: "submit_refused" },
|
||||
{ ok: false, sessionName: s, sent: true, outcome: "failed", inputPhase: "written", reason: "draft_input_changed" },
|
||||
{ ok: false, sessionName: s, sent: false, inputPhase: "not_reached" },
|
||||
{ ok: false, sessionName: s, reason: "transport_unavailable" },
|
||||
].map((result) => writtenFact(result as Parameters<typeof writtenFact>[0])))
|
||||
.toEqual(["submitted", "submitted", "submitted", "submit_only", "text_only", "none", "uncertain",
|
||||
"text_submit_unknown", "text_submit_unknown", "text_only", "none", "unknown"]);
|
||||
});
|
||||
|
||||
it("counts as admitted only an outbox row that was written, and names the message by the first outbox id", async () => {
|
||||
seedCanonicalRig();
|
||||
db.exec(outboxEntriesSchema.sql);
|
||||
const row = new OutboxHandler(db).record({ senderSession: "sender@my-rig", destinationSession: "dev-impl@my-rig", body: "hello" });
|
||||
await createTransport(typing(), { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello", { committedOutboxIds: [row.outboxId, "outbox-never-written"] });
|
||||
expect(evidence()).toMatchObject([{ admitted: { outboxIds: [row.outboxId] }, messageId: row.outboxId }]);
|
||||
});
|
||||
|
||||
it("names the occupant the send was for, even when a new one takes the seat while it runs", async () => {
|
||||
const { node } = seedCanonicalRig();
|
||||
const ordinal = (db.prepare("SELECT MAX(generation_ordinal) AS m FROM occupant_tenures WHERE node_id = ?").get(node.id) as { m: number | null }).m ?? 0;
|
||||
const tenure = db.prepare("INSERT INTO occupant_tenures (id, node_id, generation_ordinal, generation_uuid, kind) VALUES (?, ?, ?, ?, ?)");
|
||||
tenure.run("g-before", node.id, ordinal + 1, "uuid-before", "adopt");
|
||||
const handover = mockTmux({ capturePaneContent: async () => "hello", sendText: async () => {
|
||||
tenure.run("g-after", node.id, ordinal + 2, "uuid-after", "adopt");
|
||||
return { ok: true as const };
|
||||
} });
|
||||
await createTransport(handover, { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello");
|
||||
expect(evidence()).toMatchObject([{ occupantGeneration: "uuid-before" }]);
|
||||
});
|
||||
|
||||
it("an external CLI seat's refusal records nothing written and no screen check", async () => {
|
||||
seedExternalCliRig();
|
||||
await createTransport(mockTmux(), { eventBus: new EventBus(db) }).send("orch1-lead@rigged-buildout", "hello");
|
||||
expect(evidence()).toMatchObject([{ seatKind: "external_cli", written: "none", terminalObserved: { outcome: null, verified: null },
|
||||
wireSha256: null, wireBytes: null, writtenAt: null }]);
|
||||
});
|
||||
|
||||
it("a write that may have landed reads uncertain and is written once, never resent", async () => {
|
||||
seedCanonicalRig();
|
||||
let writes = 0;
|
||||
const killed = mockTmux({ sendText: async () => { writes += 1; return { ok: false, code: "tmux_error", message: "execution deadline exceeded; client killed" }; } });
|
||||
const unknown = await createTransport(killed, { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello");
|
||||
expect(unknown).toMatchObject({ outcome: "failed", inputPhase: "attempted" });
|
||||
expect(writes).toBe(1);
|
||||
expect(evidence().map((fact) => fact.written)).toEqual(["uncertain"]);
|
||||
});
|
||||
|
||||
it("text whose Enter failed or threw keeps the submission unknown; only a pre-Enter refusal reads text only", async () => {
|
||||
seedCanonicalRig();
|
||||
const send = (sendKeys: () => Promise<{ ok: false; code: string; message: string }>) =>
|
||||
createTransport(mockTmux({ capturePaneContent: async () => "hello", sendKeys }), { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello");
|
||||
expect(await send(async () => ({ ok: false, code: "tmux_error", message: "boom" }))).toMatchObject({ inputPhase: "written", reason: "submit_failed" });
|
||||
// A thrown Enter is caught on the guarded path, which reports the text written and the Enter attempted.
|
||||
const thrown = mockTmux({ capturePaneContent: async () => "hello", sendKeys: async () => { throw new Error("send-keys: lost reply"); } });
|
||||
attachGuard(thrown);
|
||||
const lost = await createTransport(thrown, { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello");
|
||||
expect(lost).toMatchObject({ inputPhase: "written" });
|
||||
expect(lost.error).toContain("the submission (Enter) outcome is unconfirmed");
|
||||
expect(await send(async () => ({ ok: false, code: "draft_input_changed", message: "Human input arrived during this delivery." })))
|
||||
.toMatchObject({ inputPhase: "written", reason: "draft_input_changed" });
|
||||
expect(evidence().map((fact) => fact.written)).toEqual(["text_submit_unknown", "text_submit_unknown", "text_only"]);
|
||||
});
|
||||
|
||||
it("an outbox read that fails leaves admission unknown, even beside a row that was read, and the send unchanged", async () => {
|
||||
seedCanonicalRig();
|
||||
db.exec(outboxEntriesSchema.sql);
|
||||
const row = new OutboxHandler(db).record({ senderSession: "sender@my-rig", destinationSession: "dev-impl@my-rig", body: "hello" });
|
||||
const real = OutboxHandler.prototype.getById;
|
||||
const read = vi.spyOn(OutboxHandler.prototype, "getById").mockImplementation(function (this: OutboxHandler, id: string) {
|
||||
if (id === "outbox-unreadable") throw new Error("database is locked");
|
||||
return real.call(this, id);
|
||||
});
|
||||
const opts = { committedOutboxIds: [row.outboxId, "outbox-unreadable"] };
|
||||
const plain = await createTransport(typing()).send("dev-impl@my-rig", "hello", opts);
|
||||
const result = await createTransport(typing(), { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello", opts);
|
||||
read.mockRestore();
|
||||
expect(result).toEqual(plain);
|
||||
expect(evidence()).toMatchObject([{ admitted: "unknown" }]);
|
||||
});
|
||||
|
||||
it("names a kind only from a binding the seat has: no session or no binding reads unknown, a binding written without a kind reads tmux", async () => {
|
||||
const { node } = seedCanonicalRig();
|
||||
const transport = () => createTransport(typing(), { eventBus: new EventBus(db) });
|
||||
await transport().send("ghost@nowhere", "hello");
|
||||
const binding = db.prepare("SELECT * FROM bindings WHERE node_id = ?").get(node.id) as Record<string, unknown>;
|
||||
db.prepare("DELETE FROM bindings WHERE node_id = ?").run(node.id);
|
||||
await transport().send("dev-impl@my-rig", "hello");
|
||||
// A binding row written the legacy way, naming no kind, takes the column's default.
|
||||
db.prepare("INSERT INTO bindings (id, node_id, tmux_session, tmux_pane) VALUES (?, ?, ?, ?)").run(binding.id, node.id, binding.tmux_session, binding.tmux_pane);
|
||||
await transport().send("dev-impl@my-rig", "hello");
|
||||
expect(evidence().map((fact) => fact.seatKind)).toEqual(["unknown", "unknown", "tmux"]);
|
||||
});
|
||||
|
||||
it("a record that can't be written leaves the send and its result untouched", async () => {
|
||||
seedCanonicalRig();
|
||||
const plain = await createTransport(typing()).send("dev-impl@my-rig", "hello");
|
||||
const broken = new EventBus(db);
|
||||
broken.emit = () => { throw new Error("disk full"); };
|
||||
const warn = vi.spyOn(console, "warn").mockImplementation(() => {});
|
||||
const result = await createTransport(typing(), { eventBus: broken }).send("dev-impl@my-rig", "hello");
|
||||
expect(result).toEqual(plain);
|
||||
expect(warn).toHaveBeenCalledTimes(1);
|
||||
warn.mockRestore();
|
||||
});
|
||||
|
||||
it("fingerprints the text the paste wrote, never the text itself: a terminal seat's exact text, not the envelope", async () => {
|
||||
seedCanonicalRig("terminal");
|
||||
let typed = "";
|
||||
const tmux = mockTmux({ capturePaneContent: async () => "hello", sendText: async (_target, text) => { typed = text; return { ok: true as const }; } });
|
||||
await createTransport(tmux, { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "env lines\nhello", { exactText: "hello" });
|
||||
const [fact] = evidence();
|
||||
expect(typed).toBe("hello");
|
||||
expect(fact).toMatchObject({ wireSha256: createHash("sha256").update("hello").digest("hex"), wireBytes: 5 });
|
||||
expect(Number.isNaN(Date.parse(fact.writtenAt))).toBe(false);
|
||||
expect(JSON.stringify(fact)).not.toContain("hello");
|
||||
});
|
||||
|
||||
it("fingerprints what was pasted, an urgent send's added line included, not the text the caller passed", async () => {
|
||||
seedCanonicalRig();
|
||||
let typed = "";
|
||||
const tmux = mockTmux({ capturePaneContent: async () => "hello", sendText: async (_target, text) => { typed = text; return { ok: true as const }; } });
|
||||
await createTransport(tmux, { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "hello", { urgent: { by: "orch@r" } });
|
||||
const sha = (text: string) => createHash("sha256").update(text).digest("hex");
|
||||
expect(typed).not.toBe("hello");
|
||||
expect(evidence()).toMatchObject([{ wireSha256: sha(typed), wireBytes: Buffer.byteLength(typed) }]);
|
||||
expect(evidence()[0].wireSha256).not.toBe(sha("hello"));
|
||||
});
|
||||
|
||||
it("a terminal seat's exact-text send, which re-enters send, records once", async () => {
|
||||
seedCanonicalRig("terminal");
|
||||
await createTransport(typing(), { eventBus: new EventBus(db) }).send("dev-impl@my-rig", "env lines\nhello", { exactText: "hello" });
|
||||
expect(evidence()).toHaveLength(1);
|
||||
});
|
||||
});
|
||||
|
||||
// Test 11: resolveSessions by rig returns running sessions
|
||||
it("resolveSessions by rig returns running sessions", async () => {
|
||||
seedCanonicalRig();
|
||||
|
||||
Reference in new issue
Block a user