mirror of
https://github.com/Ark0N/Codeman.git
synced 2026-11-14 10:39:32 +01:00
Merge pull request #694: feat(agent): COD-529 add bounded inspection and guarded submission
Conflicts resolved by keeping both sides: - CLAUDE.md (Agent CLI): #694's opening (an HTTP client, the experimental agent-session-routes behind agentCoordinationEnabled) with #708's clause on the directory spawn and its two refusals. - docs/api-reference.md (codeman agent): #708's spawn paragraph and --json shape, with #694's inspect/stage/submit in the verb list and the guarded endpoints named before "adds no route for the other verbs". - src/cli-agent.ts (@fileoverview): #708's quick-start / directory spawn routes and #694's experimental routes in one endpoint list; the two node:fs/promises imports (#708's realpath+stat, #694's open+stat) merged into one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
commit
5908cc51fd
50 files changed
+5408
-116
No files matched your search
@@ -706,6 +706,9 @@
|
||||
'New sessions for this case land in this group': '此案例的新会话将放入此分组',
|
||||
'Could not save the default session group.': '无法保存默认会话分组。',
|
||||
'Remote auto-reconnect': '远程自动重连',
|
||||
'Experimental agent coordination': '实验性智能体协作',
|
||||
"Lets agents read another session's output in bounded pages and type a prompt into a local Codex composer, then press Enter once after checking it is unchanged (the inspect, stage and submit commands). Experimental: the commands and their HTTP routes may change in any release. Off by default; while off those routes refuse every request.":
|
||||
'允许智能体分页读取另一个会话的输出(每页有上限),并把提示词输入本地 Codex 输入框,确认未被改动后只按一次 Enter(即 inspect、stage 和 submit 命令)。实验性功能:这些命令及其 HTTP 路由可能在任何版本中变化。默认关闭;关闭时这些路由拒绝所有请求。',
|
||||
'Pin tab names on the CLI': '在 CLI 上固定标签名',
|
||||
"Launch every Claude session with its tab name, so the CLI's prompt label and /resume list match your tabs. Claude then stops generating its own titles for those sessions, and a rename in Codeman holds only until the next turn. Applies from the next session start or respawn, no server restart needed.":
|
||||
'以标签名启动每个 Claude 会话,让 CLI 的提示框标签和 /resume 列表与你的标签一致。Claude 随后不再为这些会话生成自己的标题,在 Codeman 中的重命名也只保留到下一轮。从下一次会话启动或重新生成时生效,无需重启服务器。',
|
||||
|
||||
@@ -2234,6 +2234,13 @@
|
||||
</div>
|
||||
<label class="switch switch-sm"><input type="checkbox" id="appSettingsAgentInboxPersist"><span class="slider"></span></label>
|
||||
</div>
|
||||
<div class="set-row" data-search="experimental agent coordination inspect stage submit codeman agent guarded input">
|
||||
<div class="set-row-text">
|
||||
<span class="set-row-label">Experimental agent coordination</span>
|
||||
<span class="set-row-desc">Lets agents read another session's output in bounded pages and type a prompt into a local Codex composer, then press Enter once after checking it is unchanged (the inspect, stage and submit commands). Experimental: the commands and their HTTP routes may change in any release. Off by default; while off those routes refuse every request.</span>
|
||||
</div>
|
||||
<label class="switch switch-sm"><input type="checkbox" id="appSettingsAgentCoordination"><span class="slider"></span></label>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
||||
@@ -541,6 +541,8 @@ Object.assign(CodemanApp.prototype, {
|
||||
document.getElementById('appSettingsApprovalsInbox').checked = settings.approvalsInboxEnabled === true;
|
||||
// Agent inbox snapshot on disk: synced, default OFF.
|
||||
document.getElementById('appSettingsAgentInboxPersist').checked = settings.agentInboxPersistEnabled === true;
|
||||
// Experimental agent coordination (inspect/stage/submit routes): synced, default OFF.
|
||||
document.getElementById('appSettingsAgentCoordination').checked = settings.agentCoordinationEnabled === true;
|
||||
// Custom Model Endpoint Profiles: synced, default OFF. The toggle governs both
|
||||
// the Run-menu picker's generated entries and this settings panel's visibility;
|
||||
// the endpoint list itself is server state, loaded on demand below.
|
||||
@@ -2994,6 +2996,7 @@ Object.assign(CodemanApp.prototype, {
|
||||
configBackupDir: document.getElementById('appSettingsConfigBackupDir').value.trim(),
|
||||
...this._configBackupRetentionPayload(_prev),
|
||||
agentInboxPersistEnabled: document.getElementById('appSettingsAgentInboxPersist').checked,
|
||||
agentCoordinationEnabled: document.getElementById('appSettingsAgentCoordination').checked,
|
||||
readMyMindEnabled: document.getElementById('appSettingsReadMyMind').checked,
|
||||
ultracodeFloatingWindows: document.getElementById('appSettingsUltracodeFloatingWindows').checked,
|
||||
showMultiMonitorButton: document.getElementById('appSettingsShowMultiMonitorButton').checked,
|
||||
|
||||
@@ -0,0 +1,193 @@
|
||||
/**
|
||||
* @fileoverview Agent-facing bounded inspection and guarded stage/submit endpoints.
|
||||
* Ownership is checked before any disk or pane read. Native evidence comes only
|
||||
* from strict conversation attribution; all other modes get labeled snapshots.
|
||||
*
|
||||
* Experimental and opt-in: all three routes answer `403 FORBIDDEN`, naming the
|
||||
* setting, until the synced `agentCoordinationEnabled` is on (default OFF), the
|
||||
* same shape as `cliManagementEnabled` and `mcpSyncEnabled`. The gate runs first,
|
||||
* before session lookup, so a disabled install reveals nothing about its sessions.
|
||||
*/
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { Session } from '../../session.js';
|
||||
import type { SessionPort, InfraPort } from '../ports/index.js';
|
||||
import { getCli } from '../../config/cli-registry/registry.js';
|
||||
import { findExactCodexConversation, resolveEffectiveCodexHome } from '../../agent-codex-identity.js';
|
||||
import { InspectionReader, InspectionConflict, boundedUtf8Tail } from '../../agent-inspection.js';
|
||||
import {
|
||||
GuardedStageStore,
|
||||
GuardedInputConflict,
|
||||
guardedComposerChrome,
|
||||
MAY_BE_TYPED,
|
||||
type GuardedInputTarget,
|
||||
} from '../../agent-staging.js';
|
||||
import { ApiErrorCode, createErrorResponse, type ApiResponse } from '../../types.js';
|
||||
import { findSessionOrFail, getAuthUser, parseBody, readJsonConfig, SETTINGS_PATH } from '../route-helpers.js';
|
||||
import { AgentInspectQuerySchema, AgentStageSchema, AgentSubmitSchema } from '../schemas.js';
|
||||
|
||||
/** Default OFF, same shape as `readMcpSyncEnabled`: read fresh so a toggle applies at once. */
|
||||
export async function readAgentCoordinationEnabled(): Promise<boolean> {
|
||||
const settings = await readJsonConfig<Record<string, unknown>>(SETTINGS_PATH, 'settings.json', {});
|
||||
return settings.agentCoordinationEnabled === true;
|
||||
}
|
||||
|
||||
/** The refusal every route sends while the setting is off. Exported for the tests. */
|
||||
export const AGENT_COORDINATION_DISABLED_MESSAGE =
|
||||
'Experimental agent coordination (inspect, stage, submit) is disabled. Turn on "Experimental agent coordination" in Settings (agentCoordinationEnabled) and save first.';
|
||||
|
||||
async function gate(): Promise<ApiResponse<never> | null> {
|
||||
if (await readAgentCoordinationEnabled()) return null;
|
||||
return createErrorResponse(ApiErrorCode.FORBIDDEN, AGENT_COORDINATION_DISABLED_MESSAGE);
|
||||
}
|
||||
|
||||
/** A Node error code (`EACCES`, `EIO`, ...) or the error's class name; never its message. */
|
||||
export function errorCode(error: unknown): string {
|
||||
const code = (error as { code?: unknown } | null)?.code;
|
||||
if (typeof code === 'string' && /^[A-Z0-9_]{1,40}$/.test(code)) return code;
|
||||
return error instanceof Error ? error.constructor.name : 'unknown';
|
||||
}
|
||||
|
||||
export function registerAgentSessionRoutes(app: FastifyInstance, ctx: SessionPort & InfraPort): void {
|
||||
const reader = new InspectionReader();
|
||||
const stages = new GuardedStageStore();
|
||||
app.addHook('onClose', async () => {
|
||||
reader.clear();
|
||||
stages.clear();
|
||||
});
|
||||
const generation = (session: Session) => session.inspectionGeneration;
|
||||
const target = (session: Session): GuardedInputTarget => ({
|
||||
id: session.id,
|
||||
chrome: () => guardedComposerChrome(getCli(session.mode)),
|
||||
snapshot: () => {
|
||||
if (ctx.sessions.get(session.id) !== session) return null;
|
||||
const snapshot = session.guardedInputSnapshot?.();
|
||||
return snapshot ?? null;
|
||||
},
|
||||
write: (text) => session.stageGuardedInput(text),
|
||||
submit: (expected) => session.submitGuardedInput(expected),
|
||||
});
|
||||
app.get('/api/sessions/:id/inspect', async (req, reply) => {
|
||||
const refused = await gate();
|
||||
if (refused) return reply.code(403).send(refused);
|
||||
const { id } = req.params as { id: string };
|
||||
const session = findSessionOrFail(ctx, id, req);
|
||||
const { maxBytes, since } = parseBody(AgentInspectQuerySchema, req.query);
|
||||
const initialGeneration = generation(session);
|
||||
const identity = { sessionId: session.id, mode: session.mode };
|
||||
const activity = { status: session.status, lastSubmitAt: session.lastSubmitAt || null, observedAt: Date.now() };
|
||||
try {
|
||||
if (getCli(session.mode)?.capabilities.transcript === 'codex-rollout' && !session.remote && !session.docker) {
|
||||
const conversation = await findExactCodexConversation(
|
||||
{
|
||||
codexHome: resolveEffectiveCodexHome(session.getCodexHomeOverride(), session.workingDir),
|
||||
sessionId: session.id,
|
||||
workingDir: session.workingDir,
|
||||
resumeSessionId: session.codexConfig?.resumeSessionId,
|
||||
},
|
||||
{ maxEntries: 8192, maxBytes: 1048576 }
|
||||
);
|
||||
if (generation(session) !== initialGeneration || ctx.sessions.get(id) !== session)
|
||||
throw new InspectionConflict('Session changed during inspection');
|
||||
if (conversation) {
|
||||
const result = await reader.read(
|
||||
conversation.path,
|
||||
{
|
||||
owner: getAuthUser(req).username,
|
||||
sessionId: id,
|
||||
generation: initialGeneration,
|
||||
conversationId: conversation.id,
|
||||
},
|
||||
maxBytes,
|
||||
since
|
||||
);
|
||||
if (generation(session) !== initialGeneration || ctx.sessions.get(id) !== session)
|
||||
throw new InspectionConflict('Session changed during inspection');
|
||||
return {
|
||||
success: true,
|
||||
data: {
|
||||
source: 'codex-jsonl',
|
||||
identity: { ...identity, conversationId: conversation.id },
|
||||
activity,
|
||||
incremental: true,
|
||||
maxBytes,
|
||||
...result,
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
if (since)
|
||||
throw new InspectionConflict('Attributed transcript unavailable; this cursor cannot resume a pane snapshot');
|
||||
const frame = session.muxName ? ctx.mux.capturePaneText?.(session.muxName) : null;
|
||||
const evidence = frame ?? session.terminalBuffer;
|
||||
return {
|
||||
success: true,
|
||||
data: {
|
||||
source: 'pane-snapshot',
|
||||
identity,
|
||||
activity,
|
||||
incremental: false,
|
||||
cursor: null,
|
||||
maxBytes,
|
||||
evidence: boundedUtf8Tail(evidence, maxBytes),
|
||||
truncated: evidence.length > maxBytes || Buffer.byteLength(evidence) > maxBytes,
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
if (error instanceof InspectionConflict)
|
||||
return reply.code(409).send(createErrorResponse(ApiErrorCode.CONFLICT, error.message));
|
||||
// Anything else (an unreadable rollout, an IO error) is not a retryable cursor
|
||||
// conflict. Log the session and the error code only: the message can carry the
|
||||
// native transcript path, which must never reach the log or the response.
|
||||
console.warn(`[agent-inspect] session ${session.id}: inspection failed (${errorCode(error)})`);
|
||||
return reply.code(422).send(createErrorResponse(ApiErrorCode.OPERATION_FAILED, 'Inspection source unavailable'));
|
||||
}
|
||||
});
|
||||
app.post('/api/sessions/:id/stage', async (req, reply) => {
|
||||
const refused = await gate();
|
||||
if (refused) return reply.code(403).send(refused);
|
||||
const session = findSessionOrFail(ctx, (req.params as { id: string }).id, req);
|
||||
const { message } = parseBody(AgentStageSchema, req.body);
|
||||
try {
|
||||
const receipt = await stages.stage(target(session), getAuthUser(req).username, message);
|
||||
return { success: true, data: receipt };
|
||||
} catch (error) {
|
||||
if (!(error instanceof GuardedInputConflict))
|
||||
console.warn(`[agent-stage] session ${session.id}: delivery failed (${errorCode(error)})`);
|
||||
return reply
|
||||
.code(409)
|
||||
.send(
|
||||
createErrorResponse(
|
||||
ApiErrorCode.CONFLICT,
|
||||
error instanceof GuardedInputConflict
|
||||
? error.message
|
||||
: `Stage delivery failed or ambiguous; ${MAY_BE_TYPED}`
|
||||
)
|
||||
);
|
||||
}
|
||||
});
|
||||
app.post('/api/sessions/:id/submit', async (req, reply) => {
|
||||
const refused = await gate();
|
||||
if (refused) return reply.code(403).send(refused);
|
||||
const session = findSessionOrFail(ctx, (req.params as { id: string }).id, req);
|
||||
const { stageId, confirmSha } = parseBody(AgentSubmitSchema, req.body);
|
||||
try {
|
||||
return {
|
||||
success: true,
|
||||
data: await stages.submit(target(session), getAuthUser(req).username, stageId, confirmSha),
|
||||
};
|
||||
} catch (error) {
|
||||
if (!(error instanceof GuardedInputConflict))
|
||||
console.warn(`[agent-submit] session ${session.id}: delivery failed (${errorCode(error)})`);
|
||||
return reply
|
||||
.code(409)
|
||||
.send(
|
||||
createErrorResponse(
|
||||
ApiErrorCode.CONFLICT,
|
||||
error instanceof GuardedInputConflict
|
||||
? error.message
|
||||
: 'Submit delivery failed or ambiguous; receipt consumed, inspect the pane'
|
||||
)
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -47,3 +47,5 @@ export {
|
||||
type CustomModelSwapDisplacement,
|
||||
} from './custom-model-routes.js';
|
||||
export { registerCliRegistryRoutes, readCliManagementEnabled, type CliListItem } from './cli-registry-routes.js';
|
||||
|
||||
export { registerAgentSessionRoutes } from './agent-session-routes.js';
|
||||
@@ -35,6 +35,7 @@ import {
|
||||
type RemoteHost,
|
||||
} from '../../types.js';
|
||||
import {
|
||||
GuardedInputBusyError,
|
||||
Session,
|
||||
cliTakesSessionModel,
|
||||
isAltScreenStripMode,
|
||||
@@ -2078,6 +2079,32 @@ export function registerSessionRoutes(
|
||||
}
|
||||
}
|
||||
|
||||
// A guarded submit (agent-staging.ts) holds the input lease for the few
|
||||
// milliseconds of its conditional Enter, and input that arrives meanwhile is held
|
||||
// and written in arrival order after it. A TAGGED request is refused for that span
|
||||
// (the lease plus the held input's drain) rather than overtake held input: say so
|
||||
// instead of answering 200 for bytes that were never written, and give the seq
|
||||
// back so the browser's durable queue (which keeps a frame on any 409, and whose
|
||||
// POST drainer sends one frame at a time) delivers it on its next retry. An
|
||||
// UNTAGGED request has no retry behind it (the phone accessory keys and the Stop
|
||||
// button's Ctrl+C are fire-and-forget POSTs that never read the answer), so it is
|
||||
// held instead, like a WebSocket frame: `writeViaMux` takes its own turn behind
|
||||
// the lease, and the raw write below is held explicitly. Nothing from here to the
|
||||
// write below awaits, so the lease cannot be taken in between.
|
||||
const refuseGuarded = () =>
|
||||
reply
|
||||
.code(409)
|
||||
.send(
|
||||
createErrorResponse(
|
||||
ApiErrorCode.CONFLICT,
|
||||
'A guarded submit is in progress on this session; nothing was written. Retry the input.'
|
||||
)
|
||||
);
|
||||
if (!duplicate && tagged && session.guardedInputHeld === true) {
|
||||
session.forgetInputSeq(clientId as string, seq as number);
|
||||
return refuseGuarded();
|
||||
}
|
||||
|
||||
// Only a waiting request pays for the tmux probe: the browser's plain input path
|
||||
// (thousands of calls per session) must stay exec-free.
|
||||
const workerDead = wantsWait && workerIsDead(ctx.mux, session);
|
||||
@@ -2188,7 +2215,38 @@ export function registerSessionRoutes(
|
||||
// Same rollback. NOT an error response, deliberately: a session can
|
||||
// legitimately have no PTY yet (created but not started), and callers have
|
||||
// always been able to write to one without a 4xx.
|
||||
delivered = session.write(inputStr, { fromUser: true });
|
||||
//
|
||||
// During a guarded submit (untagged only: a tagged one was refused above) the
|
||||
// write is held and runs in arrival order once the lease is released. The answer
|
||||
// is a 200 once held; a send-and-wait reads `delivered` from the held write. Only
|
||||
// a full hold queue refuses, after releasing the waiter registered above.
|
||||
let heldWrite: Promise<boolean> | null = null;
|
||||
if (session.guardedInputHeld === true) {
|
||||
let settle!: (written: boolean) => void;
|
||||
const written = new Promise<boolean>((resolve) => {
|
||||
settle = resolve;
|
||||
});
|
||||
const held = session.holdForGuardedLease(
|
||||
() => {
|
||||
let ok = false;
|
||||
try {
|
||||
ok = session.write(inputStr, { fromUser: true });
|
||||
} finally {
|
||||
settle(ok);
|
||||
}
|
||||
},
|
||||
inputStr.length,
|
||||
inputStr
|
||||
);
|
||||
if (held === 'full') {
|
||||
abort.abort();
|
||||
stopDeathWatch();
|
||||
return refuseGuarded();
|
||||
}
|
||||
if (held === 'held') heldWrite = written;
|
||||
}
|
||||
if (heldWrite && !waitPromise) return {};
|
||||
delivered = heldWrite ? await heldWrite : session.write(inputStr, { fromUser: true });
|
||||
if (!delivered && tagged) {
|
||||
session.forgetInputSeq(clientId as string, seq as number);
|
||||
}
|
||||
@@ -2399,7 +2457,7 @@ export function registerSessionRoutes(
|
||||
// own `capabilities.newline`. Claude Code's Ink input recognizes 0x0a as "insert newline" vs
|
||||
// 0x0d (carriage return = submit).
|
||||
|
||||
app.post('/api/sessions/:id/send-key', async (req) => {
|
||||
app.post('/api/sessions/:id/send-key', async (req, reply) => {
|
||||
const { id } = req.params as { id: string };
|
||||
const body = req.body as Record<string, unknown>;
|
||||
const key = typeof body?.key === 'string' ? body.key : '';
|
||||
@@ -2429,18 +2487,26 @@ export function registerSessionRoutes(
|
||||
try {
|
||||
// Route through the dedicated Codeman socket — bare `tmux` would target the
|
||||
// user's default server and never find this session (same #80 regression class).
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
execFile(
|
||||
'tmux',
|
||||
['-L', ctx.mux.muxSocket, 'send-keys', '-H', '-t', muxName, ...hex],
|
||||
{ timeout: 5000 },
|
||||
(err) => {
|
||||
if (err) reject(err);
|
||||
else resolve();
|
||||
}
|
||||
);
|
||||
});
|
||||
await session.performAuxiliaryInput(
|
||||
() =>
|
||||
new Promise<void>((resolve, reject) => {
|
||||
execFile(
|
||||
'tmux',
|
||||
['-L', ctx.mux.muxSocket, 'send-keys', '-H', '-t', muxName, ...hex],
|
||||
{ timeout: 5000 },
|
||||
(err) => {
|
||||
if (err) reject(err);
|
||||
else resolve();
|
||||
}
|
||||
);
|
||||
})
|
||||
);
|
||||
} catch (err) {
|
||||
// A guarded submit's lease holds the key until its Enter is out; only a full
|
||||
// hold queue refuses it, and that is a retryable conflict, not a tmux failure.
|
||||
if (err instanceof GuardedInputBusyError) {
|
||||
return reply.code(409).send(createErrorResponse(ApiErrorCode.CONFLICT, err.message));
|
||||
}
|
||||
console.error('[Server] send-key failed:', err);
|
||||
return createErrorResponse(ApiErrorCode.INTERNAL_ERROR, 'tmux send-keys failed');
|
||||
}
|
||||
|
||||
+83
-45
@@ -29,6 +29,11 @@
|
||||
* {"t":"i","d":"...","seq":N,"cid":"..."} — input (keystroke or paste). seq+cid are
|
||||
* optional reliable-delivery tags: the server applies each
|
||||
* (cid,seq) at-most-once and ACKs with {"t":"ia","seq":N}.
|
||||
* A frame that arrives while a guarded submit holds the
|
||||
* session's input lease is held, then applied and ACKed in
|
||||
* arrival order once it is released; when that hold queue
|
||||
* is full the socket is closed with 1013 and nothing later
|
||||
* from it is applied, so the client resends in order.
|
||||
* {"t":"z","c":N,"r":N,"f":bool} — resize terminal (f=true forces SIGWINCH even if dims unchanged)
|
||||
*/
|
||||
|
||||
@@ -169,12 +174,69 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost
|
||||
const sizingToken = Symbol('ws-desktop-sizing');
|
||||
let holdsDesktopClaim = false;
|
||||
|
||||
// Set once an input frame could not be held behind a guarded submit's lease
|
||||
// (the hold queue was full). Nothing later from this socket is applied: it is
|
||||
// closed, and the browser reconnects and resends everything unACKed, in order.
|
||||
let inputStalled = false;
|
||||
|
||||
// Apply one input frame and ACK it. Runs inline, or later and in arrival order
|
||||
// when a guarded submit held the input lease as the frame arrived.
|
||||
const applyInputFrame = (msg: { d: string; seq?: unknown; cid?: unknown }) => {
|
||||
// Reliable delivery: when the frame carries a clientId + seq, apply it
|
||||
// exactly once (skip a duplicate redelivery) but ACK it regardless so
|
||||
// the client can drop it from its durable queue. Frames without seq
|
||||
// (legacy/other tools) are applied as-is — no behavior change.
|
||||
const cid = typeof msg.cid === 'string' ? msg.cid : null;
|
||||
const seq = Number.isInteger(msg.seq) ? (msg.seq as number) : null;
|
||||
const apply = cid && seq !== null ? session.shouldApplyInput(cid, seq) : true;
|
||||
let delivered = true;
|
||||
if (apply) {
|
||||
// Typed input from a claim-holding desktop keeps the claim "hot"
|
||||
// and re-asserts the desktop layout after a mobile override.
|
||||
if (holdsDesktopClaim) session.noteDesktopActivity();
|
||||
// Browser keystrokes are the user's own, so they may name the tab.
|
||||
delivered = session.write(msg.d, { fromUser: true });
|
||||
// A session whose PTY is gone swallows the write. ACKing anyway told
|
||||
// the client to drop the frame from its durable queue and left the seq
|
||||
// burnt, so the retry that reliable delivery exists for was rejected as
|
||||
// a duplicate: the input was lost for good.
|
||||
if (!delivered && cid && seq !== null) session.forgetInputSeq(cid, seq);
|
||||
}
|
||||
if (seq !== null && socket.readyState === 1) {
|
||||
if (apply) {
|
||||
if (delivered) socket.send(`{"t":"ia","seq":${seq}}`);
|
||||
} else {
|
||||
// REJECTED as a duplicate. ACK it — the client must still drop it
|
||||
// from its durable queue — but say so, and hand back our watermark.
|
||||
//
|
||||
// A plain ACK here is indistinguishable from "applied", which is
|
||||
// what made a client with a rolled-back counter unrecoverable: its
|
||||
// seqs persist to localStorage on a DEBOUNCED write, so a tab killed
|
||||
// between a send and that write comes back with a counter BELOW this
|
||||
// watermark, every later keystroke lands at or under it, and each one
|
||||
// is dropped-but-ACKed. The UI stays clean, nothing is delivered, and
|
||||
// a reload restores the same stale counter. `last` is what lets the
|
||||
// client lift itself out.
|
||||
// ⚠️ Defensive: the session arrives through a structural port, and an
|
||||
// implementation without this method must not take the whole input
|
||||
// path down with it — a throw here aborts the message handler and the
|
||||
// frame is never ACKed at all, which strands it in the client's queue.
|
||||
const watermark =
|
||||
typeof (session as { lastInputSeq?: (c: string) => number }).lastInputSeq === 'function'
|
||||
? (session as { lastInputSeq: (c: string) => number }).lastInputSeq(cid as string)
|
||||
: seq;
|
||||
socket.send(`{"t":"ia","seq":${seq},"dup":true,"last":${watermark}}`);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Attach message handler synchronously BEFORE any async work
|
||||
// (@fastify/websocket requirement to avoid dropped messages).
|
||||
socket.on('message', (raw) => {
|
||||
try {
|
||||
const msg = JSON.parse(String(raw));
|
||||
if (msg.t === 'i' && typeof msg.d === 'string') {
|
||||
if (inputStalled) return;
|
||||
if (msg.d.length > MAX_INPUT_LENGTH) {
|
||||
// Refused for good, so say so: a silent return left the frame
|
||||
// unACKed and the client redelivered it every few seconds forever
|
||||
@@ -185,51 +247,27 @@ export function registerWsRoutes(app: FastifyInstance, ctx: SessionPort, getHost
|
||||
}
|
||||
return;
|
||||
}
|
||||
// Reliable delivery: when the frame carries a clientId + seq, apply it
|
||||
// exactly once (skip a duplicate redelivery) but ACK it regardless so
|
||||
// the client can drop it from its durable queue. Frames without seq
|
||||
// (legacy/other tools) are applied as-is — no behavior change.
|
||||
const cid = typeof msg.cid === 'string' ? msg.cid : null;
|
||||
const seq = Number.isInteger(msg.seq) ? (msg.seq as number) : null;
|
||||
const apply = cid && seq !== null ? session.shouldApplyInput(cid, seq) : true;
|
||||
let delivered = true;
|
||||
if (apply) {
|
||||
// Typed input from a claim-holding desktop keeps the claim "hot"
|
||||
// and re-asserts the desktop layout after a mobile override.
|
||||
if (holdsDesktopClaim) session.noteDesktopActivity();
|
||||
// Browser keystrokes are the user's own, so they may name the tab.
|
||||
delivered = session.write(msg.d, { fromUser: true });
|
||||
// A session whose PTY is gone swallows the write. ACKing anyway told
|
||||
// the client to drop the frame from its durable queue and left the seq
|
||||
// burnt, so the retry that reliable delivery exists for was rejected as
|
||||
// a duplicate: the input was lost for good.
|
||||
if (!delivered && cid && seq !== null) session.forgetInputSeq(cid, seq);
|
||||
}
|
||||
if (seq !== null && socket.readyState === 1) {
|
||||
if (apply) {
|
||||
if (delivered) socket.send(`{"t":"ia","seq":${seq}}`);
|
||||
} else {
|
||||
// REJECTED as a duplicate. ACK it — the client must still drop it
|
||||
// from its durable queue — but say so, and hand back our watermark.
|
||||
//
|
||||
// A plain ACK here is indistinguishable from "applied", which is
|
||||
// what made a client with a rolled-back counter unrecoverable: its
|
||||
// seqs persist to localStorage on a DEBOUNCED write, so a tab killed
|
||||
// between a send and that write comes back with a counter BELOW this
|
||||
// watermark, every later keystroke lands at or under it, and each one
|
||||
// is dropped-but-ACKed. The UI stays clean, nothing is delivered, and
|
||||
// a reload restores the same stale counter. `last` is what lets the
|
||||
// client lift itself out.
|
||||
// ⚠️ Defensive: the session arrives through a structural port, and an
|
||||
// implementation without this method must not take the whole input
|
||||
// path down with it — a throw here aborts the message handler and the
|
||||
// frame is never ACKed at all, which strands it in the client's queue.
|
||||
const watermark =
|
||||
typeof (session as { lastInputSeq?: (c: string) => number }).lastInputSeq === 'function'
|
||||
? (session as { lastInputSeq: (c: string) => number }).lastInputSeq(cid as string)
|
||||
: seq;
|
||||
socket.send(`{"t":"ia","seq":${seq},"dup":true,"last":${watermark}}`);
|
||||
}
|
||||
// A guarded submit (agent-staging.ts) holds the input lease across its
|
||||
// awaited Enter, so frames keep arriving while it is held. Refusing one
|
||||
// lost it: this path does not wait for ACKs, so the next seq landed after
|
||||
// the lease, moved the watermark past the refused one, and its redelivery
|
||||
// was then dropped as a duplicate. Hold the frame instead; the session runs
|
||||
// it, ACK included, in arrival order the moment the lease is released.
|
||||
// ⚠️ Defensive, like `lastInputSeq` below: the session arrives through a
|
||||
// structural port, and one without this method must still deliver input
|
||||
// (a throw here would drop the frame unACKed).
|
||||
const held =
|
||||
typeof session.holdForGuardedLease === 'function'
|
||||
? session.holdForGuardedLease(() => applyInputFrame(msg), msg.d.length, msg.d)
|
||||
: 'free';
|
||||
if (held === 'free') {
|
||||
applyInputFrame(msg);
|
||||
} else if (held === 'full') {
|
||||
// Not held, not applied, not ACKed, and no later frame from this socket
|
||||
// is applied either, so nothing can overtake it.
|
||||
inputStalled = true;
|
||||
console.warn('[ws] input backlog during a guarded submit; closing for redelivery', { sessionId: id });
|
||||
socket.close(1013, 'Input backlog during guarded submit');
|
||||
}
|
||||
} else if (
|
||||
msg.t === 'z' &&
|
||||
|
||||
@@ -1557,6 +1557,13 @@ export const SettingsUpdateSchema = z
|
||||
* inboxes stay). On, pending messages between agents survive a server restart.
|
||||
*/
|
||||
agentInboxPersistEnabled: z.boolean().optional(),
|
||||
/**
|
||||
* Experimental agent coordination (src/web/routes/agent-session-routes.ts): bounded
|
||||
* inspection and the guarded stage/submit routes behind `codeman agent inspect|stage|submit`.
|
||||
* SYNCED, default OFF: the routes are labeled experimental and may change in a minor or
|
||||
* patch release, so automation has to opt in. While OFF, all three routes answer 403.
|
||||
*/
|
||||
agentCoordinationEnabled: z.boolean().optional(),
|
||||
/**
|
||||
* Read My Mind predictor model override. Empty/absent = the AI-checker
|
||||
* default (opus: prediction quality is the product and it runs only on an
|
||||
@@ -2432,3 +2439,15 @@ export const CustomModelSelectionSchema = z.union([
|
||||
}),
|
||||
z.object({ clear: z.literal(true) }),
|
||||
]);
|
||||
|
||||
/** Agent inspection bounds are bytes of evidence, independent of the JSON envelope. */
|
||||
export const AgentInspectQuerySchema = z
|
||||
.object({
|
||||
maxBytes: z.coerce.number().int().min(1).max(65536).default(8192),
|
||||
since: z.string().uuid().optional(),
|
||||
})
|
||||
.strict();
|
||||
export const AgentStageSchema = z.object({ message: z.string().min(1).max(8192) }).strict();
|
||||
export const AgentSubmitSchema = z
|
||||
.object({ stageId: z.string().uuid(), confirmSha: z.string().regex(/^[a-f0-9]{64}$/) })
|
||||
.strict();
|
||||
@@ -219,6 +219,7 @@ import {
|
||||
registerCaseRoutes,
|
||||
registerWorktreeRoutes,
|
||||
registerSessionRoutes,
|
||||
registerAgentSessionRoutes,
|
||||
registerRespawnRoutes,
|
||||
registerRalphRoutes,
|
||||
registerPlanRoutes,
|
||||
@@ -1306,6 +1307,7 @@ export class WebServer extends EventEmitter {
|
||||
// The registry's lifetime is the server's: it drops per-session wake state on every
|
||||
// cleanup path and resolves in-flight wakes on shutdown.
|
||||
this.remoteWake = registerSessionRoutes(this.app, ctx);
|
||||
registerAgentSessionRoutes(this.app, ctx);
|
||||
registerRespawnRoutes(this.app, ctx);
|
||||
registerRalphRoutes(this.app, ctx);
|
||||
registerPlanRoutes(this.app, ctx);
|
||||
|
||||
Reference in new issue
Block a user