fix(input): fiabilise la livraison de délégation et journalise le submit
Durcit le portail d'écriture de délégation et son acheminement bout-en-bout (commande Tauri → orchestrateur → file d'entrée infrastructure → portail frontend), avec une journalisation du submit pour diagnostiquer les cas où la délégation n'était pas remise. Couvre le portail d'écriture côté front (useWritePortal) avec ses tests. QA : checks application/front/infrastructure/app-tauri verts. Les tests loopback socket Unix réels ne sont pas exécutables en sandbox (UnixListener::bind → PermissionDenied) ; alternatives avec skips vertes. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -28,15 +28,27 @@ export class TauriInputGateway implements InputGateway {
|
||||
agentId: string,
|
||||
ticket: string,
|
||||
): Promise<void> {
|
||||
console.info("[input-gateway] delegationDelivered:start", {
|
||||
projectId,
|
||||
agentId,
|
||||
ticket,
|
||||
});
|
||||
await invoke("delegation_delivered", {
|
||||
request: { projectId, agentId, ticket },
|
||||
});
|
||||
console.info("[input-gateway] delegationDelivered:ok", {
|
||||
projectId,
|
||||
agentId,
|
||||
ticket,
|
||||
});
|
||||
}
|
||||
|
||||
async setFrontAttached(agentId: string, attached: boolean): Promise<void> {
|
||||
console.info("[input-gateway] setFrontAttached:start", { agentId, attached });
|
||||
await invoke("set_front_attached", {
|
||||
request: { agentId, attached },
|
||||
});
|
||||
console.info("[input-gateway] setFrontAttached:ok", { agentId, attached });
|
||||
}
|
||||
|
||||
async cancelResume(agentId: string): Promise<boolean> {
|
||||
|
||||
@ -64,11 +64,13 @@ export function makeTerminalHandle(
|
||||
return {
|
||||
sessionId,
|
||||
write(data: Uint8Array): Promise<void> {
|
||||
const run = chain.then(() =>
|
||||
invoke<void>("write_terminal", {
|
||||
const run = chain.then(async () => {
|
||||
logTerminalWrite("start", sessionId, data);
|
||||
await invoke<void>("write_terminal", {
|
||||
request: { sessionId, data: Array.from(data) },
|
||||
}),
|
||||
);
|
||||
});
|
||||
logTerminalWrite("ok", sessionId, data);
|
||||
});
|
||||
// Keep the chain alive even if this write rejects: the next write must
|
||||
// still run. Swallow the error on the *chain* copy only — `run` keeps the
|
||||
// rejection so the caller can observe it.
|
||||
@ -91,6 +93,35 @@ export function makeTerminalHandle(
|
||||
};
|
||||
}
|
||||
|
||||
function logTerminalWrite(phase: "start" | "ok", sessionId: string, data: Uint8Array): void {
|
||||
if (data.length === 1 && data[0] >= 0x20 && data[0] !== 0x7f) return;
|
||||
console.info("[terminal-write]", phase, {
|
||||
sessionId,
|
||||
bytes: data.length,
|
||||
control: describeBytes(data),
|
||||
});
|
||||
}
|
||||
|
||||
function describeBytes(data: Uint8Array): string {
|
||||
if (data.length === 0) return "<empty>";
|
||||
if (data.length > 16) return "<bulk>";
|
||||
return Array.from(data)
|
||||
.map((byte) => {
|
||||
switch (byte) {
|
||||
case 0x0d:
|
||||
return "\\r";
|
||||
case 0x0a:
|
||||
return "\\n";
|
||||
case 0x7f:
|
||||
return "\\x7f";
|
||||
default:
|
||||
if (byte < 0x20) return `\\x${byte.toString(16).padStart(2, "0")}`;
|
||||
return String.fromCharCode(byte);
|
||||
}
|
||||
})
|
||||
.join("");
|
||||
}
|
||||
|
||||
export class TauriTerminalGateway implements TerminalGateway {
|
||||
async openTerminal(
|
||||
options: OpenTerminalOptions,
|
||||
|
||||
@ -27,7 +27,7 @@ export type DomainEvent =
|
||||
* PTY-writes the turn: the frontend write-portal runs the handshake (b→e)
|
||||
* and writes `text` + `submitSequence`. `submitSequence`/`submitDelayMs`
|
||||
* come from the target's profile; absent ⇒ the portal applies its defaults
|
||||
* (`"\r"`, ~60 ms).
|
||||
* (`"\r"`, ~350 ms).
|
||||
*/
|
||||
type: "delegationReady";
|
||||
agentId: string;
|
||||
|
||||
@ -235,6 +235,26 @@ describe("useWritePortal (§20)", () => {
|
||||
expect(writes).toEqual(["msg", "\n"]);
|
||||
});
|
||||
|
||||
it("defaults an empty profile submitSequence to Enter", async () => {
|
||||
const { system, view } = setup();
|
||||
const { handle, writes } = makeHandle();
|
||||
act(() => view.result.current.portal.bindHandle(handle));
|
||||
|
||||
await act(async () => {
|
||||
system.emit({
|
||||
type: "delegationReady",
|
||||
agentId: "ag1",
|
||||
ticket: "t1",
|
||||
text: "msg",
|
||||
submitSequence: "",
|
||||
submitDelayMs: 0,
|
||||
});
|
||||
await vi.runAllTimersAsync();
|
||||
});
|
||||
|
||||
expect(writes).toEqual(["msg", "\r"]);
|
||||
});
|
||||
|
||||
it("ignores delegationReady for a different agent", async () => {
|
||||
const { system, view } = setup();
|
||||
const { handle, writes } = makeHandle();
|
||||
|
||||
@ -58,6 +58,8 @@ const DELEGATION_WRITE_CHUNK_BYTES = 512;
|
||||
/** Delay between delegation chunks, to avoid TUI paste/drop behaviour on cold agents. */
|
||||
const DELEGATION_WRITE_CHUNK_DELAY_MS = 8;
|
||||
|
||||
const LOG_PREFIX = "[write-portal]";
|
||||
|
||||
/** A pending delegation kept in the local FIFO. */
|
||||
interface PendingDelegation {
|
||||
ticket: string;
|
||||
@ -135,10 +137,24 @@ export function useWritePortal(
|
||||
if (previous === desired) return;
|
||||
|
||||
if (previous) {
|
||||
void inputRef.current?.setFrontAttached(previous, false).catch(() => {});
|
||||
portalLog("front-attached:set", { agent: previous, attached: false });
|
||||
void inputRef.current?.setFrontAttached(previous, false).catch((error) => {
|
||||
portalLog("front-attached:set-failed", {
|
||||
agent: previous,
|
||||
attached: false,
|
||||
error: describeError(error),
|
||||
});
|
||||
});
|
||||
}
|
||||
if (desired) {
|
||||
void inputRef.current?.setFrontAttached(desired, true).catch(() => {});
|
||||
portalLog("front-attached:set", { agent: desired, attached: true });
|
||||
void inputRef.current?.setFrontAttached(desired, true).catch((error) => {
|
||||
portalLog("front-attached:set-failed", {
|
||||
agent: desired,
|
||||
attached: true,
|
||||
error: describeError(error),
|
||||
});
|
||||
});
|
||||
}
|
||||
frontAttachedAgentRef.current = desired;
|
||||
};
|
||||
@ -164,7 +180,16 @@ export function useWritePortal(
|
||||
setOverlay(true);
|
||||
|
||||
void (async () => {
|
||||
let drainNext = false;
|
||||
try {
|
||||
portalLog("inject:start", {
|
||||
agent,
|
||||
ticket: head.ticket,
|
||||
textBytes: encoder.encode(head.text).byteLength,
|
||||
queue: queueRef.current.length,
|
||||
submitSequence: describeControl(head.submitSequence),
|
||||
submitDelayMs: head.submitDelayMs ?? DEFAULT_SUBMIT_DELAY_MS,
|
||||
});
|
||||
// Yield once so any keystroke that landed in the micro-window between
|
||||
// "empty observed" (a) and the suspension taking hold (b) is counted
|
||||
// before we re-check K at step (c). Without this the race window would
|
||||
@ -174,6 +199,7 @@ export function useWritePortal(
|
||||
// (backspace) — never Ctrl-U (which could clear more than the human typed).
|
||||
const k = counterRef.current;
|
||||
if (k > 0) {
|
||||
portalLog("inject:erase-human-window", { agent, ticket: head.ticket, count: k });
|
||||
await handle.write(encoder.encode("\x7f".repeat(k)));
|
||||
counterRef.current = 0;
|
||||
}
|
||||
@ -183,19 +209,37 @@ export function useWritePortal(
|
||||
// Long delegated prompts are chunked: Codex's TUI can otherwise treat a
|
||||
// large single write like a paste and occasionally keep/drop the tail on
|
||||
// cold-started agents before Enter is received.
|
||||
await writeDelegationText(handle, head.text);
|
||||
await writeDelegationText(handle, head.text, head.ticket);
|
||||
const delay = head.submitDelayMs ?? DEFAULT_SUBMIT_DELAY_MS;
|
||||
portalLog("inject:before-submit-delay", { agent, ticket: head.ticket, delay });
|
||||
await sleep(delay);
|
||||
const submit = head.submitSequence ?? DEFAULT_SUBMIT_SEQUENCE;
|
||||
const submit = normalizeSubmitSequence(head.submitSequence, head.ticket);
|
||||
portalLog("inject:submit-write:start", {
|
||||
agent,
|
||||
ticket: head.ticket,
|
||||
submitSequence: describeControl(submit),
|
||||
submitBytes: encoder.encode(submit).byteLength,
|
||||
});
|
||||
await handle.write(encoder.encode(submit));
|
||||
portalLog("inject:submit-write:ok", { agent, ticket: head.ticket });
|
||||
|
||||
// Dequeue + ack exactly once (best-effort; never throws the handshake).
|
||||
queueRef.current.shift();
|
||||
drainNext = true;
|
||||
try {
|
||||
portalLog("inject:ack:start", { agent, ticket: head.ticket });
|
||||
await inputRef.current.delegationDelivered(project, agent, head.ticket);
|
||||
portalLog("inject:ack:ok", { agent, ticket: head.ticket });
|
||||
} catch {
|
||||
portalLog("inject:ack:failed", { agent, ticket: head.ticket });
|
||||
/* ack is observability-only; never block the portal */
|
||||
}
|
||||
} catch (error) {
|
||||
portalLog("inject:failed", {
|
||||
agent,
|
||||
ticket: head.ticket,
|
||||
error: describeError(error),
|
||||
});
|
||||
} finally {
|
||||
// (e) lower the overlay + resume the relay, with the 2 s anti-flash floor.
|
||||
const elapsed = Date.now() - startedAt;
|
||||
@ -205,7 +249,7 @@ export function useWritePortal(
|
||||
setOverlay(false);
|
||||
injectingRef.current = false;
|
||||
// A boundary may now be open for the next queued delegation.
|
||||
if (queueRef.current.length > 0) tryInject.current();
|
||||
if (queueRef.current.length > 0 && drainNext) tryInject.current();
|
||||
}
|
||||
})();
|
||||
};
|
||||
@ -231,6 +275,14 @@ export function useWritePortal(
|
||||
submitSequence: event.submitSequence,
|
||||
submitDelayMs: event.submitDelayMs,
|
||||
});
|
||||
portalLog("event:delegation-ready", {
|
||||
agent: subscribedAgent,
|
||||
ticket: event.ticket,
|
||||
textBytes: encoder.encode(event.text).byteLength,
|
||||
queue: queueRef.current.length,
|
||||
submitSequence: describeControl(event.submitSequence),
|
||||
submitDelayMs: event.submitDelayMs,
|
||||
});
|
||||
// A delegation may already be at a clean boundary (line empty) — try now.
|
||||
tryInject.current();
|
||||
})
|
||||
@ -239,6 +291,7 @@ export function useWritePortal(
|
||||
else {
|
||||
unsubscribe = un;
|
||||
subscribedAgentRef.current = subscribedAgent;
|
||||
portalLog("event:subscribed", { agent: subscribedAgent });
|
||||
reconcileFrontAttachment.current();
|
||||
}
|
||||
});
|
||||
@ -248,6 +301,7 @@ export function useWritePortal(
|
||||
unsubscribe?.();
|
||||
if (subscribedAgentRef.current === subscribedAgent) {
|
||||
subscribedAgentRef.current = null;
|
||||
portalLog("event:unsubscribed", { agent: subscribedAgent });
|
||||
}
|
||||
if (frontAttachedAgentRef.current === subscribedAgent) {
|
||||
void inputRef.current?.setFrontAttached(subscribedAgent, false).catch(() => {});
|
||||
@ -279,6 +333,7 @@ export function useWritePortal(
|
||||
},
|
||||
bindHandle(handle: TerminalHandle) {
|
||||
handleRef.current = handle;
|
||||
portalLog("handle:bound", { agent: agentIdRef.current, sessionId: handle.sessionId });
|
||||
// Tell the backend a frontend cell is mounted only after the event
|
||||
// subscription is ready. Until then, the backend keeps the headless PTY
|
||||
// write fallback, so a delegation cannot disappear during mount.
|
||||
@ -287,6 +342,10 @@ export function useWritePortal(
|
||||
tryInject.current();
|
||||
},
|
||||
unbindHandle() {
|
||||
portalLog("handle:unbound", {
|
||||
agent: agentIdRef.current,
|
||||
sessionId: handleRef.current?.sessionId,
|
||||
});
|
||||
handleRef.current = null;
|
||||
// The cell is gone — let the backend fall back to headless delivery so a
|
||||
// delegation arriving while this agent has no live cell is not lost.
|
||||
@ -307,16 +366,40 @@ function sleep(ms: number): Promise<void> {
|
||||
async function writeDelegationText(
|
||||
handle: TerminalHandle,
|
||||
text: string,
|
||||
ticket: string,
|
||||
): Promise<void> {
|
||||
const chunks = splitUtf8Chunks(text, DELEGATION_WRITE_CHUNK_BYTES);
|
||||
for (let i = 0; i < chunks.length; i += 1) {
|
||||
portalLog("inject:text-chunk:start", {
|
||||
ticket,
|
||||
sessionId: handle.sessionId,
|
||||
chunk: i + 1,
|
||||
chunks: chunks.length,
|
||||
bytes: encoder.encode(chunks[i]).byteLength,
|
||||
});
|
||||
await handle.write(encoder.encode(chunks[i]));
|
||||
portalLog("inject:text-chunk:ok", {
|
||||
ticket,
|
||||
sessionId: handle.sessionId,
|
||||
chunk: i + 1,
|
||||
chunks: chunks.length,
|
||||
});
|
||||
if (i + 1 < chunks.length) {
|
||||
await sleep(DELEGATION_WRITE_CHUNK_DELAY_MS);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeSubmitSequence(sequence: string | undefined, ticket: string): string {
|
||||
if (sequence == null) return DEFAULT_SUBMIT_SEQUENCE;
|
||||
if (sequence.length > 0) return sequence;
|
||||
portalLog("inject:empty-submit-sequence-defaulted", {
|
||||
ticket,
|
||||
defaultSubmitSequence: describeControl(DEFAULT_SUBMIT_SEQUENCE),
|
||||
});
|
||||
return DEFAULT_SUBMIT_SEQUENCE;
|
||||
}
|
||||
|
||||
function splitUtf8Chunks(text: string, maxBytes: number): string[] {
|
||||
if (text.length === 0) return [""];
|
||||
const chunks: string[] = [];
|
||||
@ -337,3 +420,36 @@ function splitUtf8Chunks(text: string, maxBytes: number): string[] {
|
||||
if (current.length > 0) chunks.push(current);
|
||||
return chunks;
|
||||
}
|
||||
|
||||
function describeControl(value: string | undefined): string {
|
||||
if (value == null) return "<default>";
|
||||
if (value.length === 0) return "<empty>";
|
||||
return [...value]
|
||||
.map((ch) => {
|
||||
switch (ch) {
|
||||
case "\r":
|
||||
return "\\r";
|
||||
case "\n":
|
||||
return "\\n";
|
||||
case "\x7f":
|
||||
return "\\x7f";
|
||||
default: {
|
||||
const code = ch.codePointAt(0)!;
|
||||
if (code < 0x20) return `\\x${code.toString(16).padStart(2, "0")}`;
|
||||
return ch;
|
||||
}
|
||||
}
|
||||
})
|
||||
.join("");
|
||||
}
|
||||
|
||||
function describeError(error: unknown): string {
|
||||
if (error && typeof error === "object" && "message" in error) {
|
||||
return String((error as { message: unknown }).message);
|
||||
}
|
||||
return String(error);
|
||||
}
|
||||
|
||||
function portalLog(event: string, fields: Record<string, unknown>): void {
|
||||
console.info(LOG_PREFIX, event, fields);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user