diff --git a/.ai/contexts/trigger-watcher.md b/.ai/contexts/trigger-watcher.md index 7bc4a4df..b6b13aaf 100644 --- a/.ai/contexts/trigger-watcher.md +++ b/.ai/contexts/trigger-watcher.md @@ -260,10 +260,112 @@ Result written to `SWITCHBOARD_TRIGGERS_DIR/processed/.result.json`: Trigger file is **deleted** after processing (success or failure). +### Removing the entry + +`writeResult()` is the single place a trigger's fate is decided: it writes the +result atomically (`.tmp` + rename) and then unlinks the trigger. Every `return` +in `processTriggerFile()` that produces a result goes through it. The body of +`processTriggerFile()` (everything past the `*.json` filename check) also runs +inside one `try`/`catch`: any exception raised anywhere in it — shape +validation, session lookup, the PTY write, a step in a `chain` — is caught and +turned into `writeResult({ ok: false, error: 'internal error: ' + err.message, +internal: true })` before the function returns. `internal: true` is set on +this path only, so a caller can tell "our code broke" apart from a validation +refusal without parsing `error` text (see `docs/automation.md`, "Reading a +result"). Between the two, there is no "processed but left behind" path — the +trigger directory lists exactly what is still pending. +Two review findings on the previous version of this file (a `null` trigger +body, and a `chain` step that is not an object) both threw *before* any +`writeResult()` call was reached; the wrapping `try`/`catch` is what closes +that gap, rather than validating every field defensively before use. + +That outer `try`/`catch`, however, only catches a throw from the *synchronous* +call into `processTriggerFile()`'s body. `waitForComposerFree`, +`pollForBusyRise`, `waitForBusyFall` and `waitForIdle` all poll on a +recursive `setTimeout` — every tick after the first runs from inside a timer +callback, a stack frame the outer `try`/`catch` never sees. All four share one +`pollLoop(check)` helper whose `tick()` wraps every call to `check` (first and +all later ones) in its own `try`/`catch` and routes a throw to that promise's +`reject` — which the `await` sites inside `processTriggerFile()` then hand to +the outer `try`/`catch` like any other exception. Before this, a throw from +one of these ctx hooks on the second tick or later had nothing to catch it: +not the original `Promise` executor (already returned), not +`processTriggerFile()`'s `try`/`catch`, not `dispatch()`'s `.catch()` — it +surfaced as an `uncaughtException` on the whole process. See +`pollLoop` in `trigger-watcher.js`. + +`writeResult()` itself is written to never throw, full stop — including when +`ctx.log.error` (supplied by the caller, out of this module's control) itself +throws. Both of its `try`/`catch` blocks route their own logging through +`safeLogError()`, which swallows whatever the logger throws, and the unlink +branch calls `onEntryRetained()` *before* attempting to log — a guarantee this +module makes must not depend on whether the log call that merely describes it +succeeds. + +Two consequences worth knowing: + +- **`ENOENT` on the unlink is not a failure.** Two `rename` events for the same + file can both reach processing; the loser finds the file already gone. That is + the intended end state, so it stays silent. The same holds for the initial + `lstat`: `ENOENT` there means the file vanished before it could be inspected, + and returns without writing a result — there is nothing to report. +- **Any other unlink error marks the name `retained`.** This is the only path + that still adds to `retained` in the running process. The entry could not be + removed, so a later filesystem event on that name would re-run a command that + already ran. `retained` (a `Set` in `start()`) makes the watcher ignore the + name for the process's lifetime, and the failure is logged at error level + rather than swallowed. This trades "processed at least once" for "never + processed twice", which is the direction the transport must fail in: the + result file is already written, so nothing is lost by refusing to look at the + leftover again. The `Set` only ever holds names whose removal failed, so it + does not grow in normal operation. + +A non-`ENOENT` `lstat` error no longer returns silently either: it goes through +`writeResult()` like everything else, with `error: 'trigger could not be +inspected: ' + err.message`. It does not call into `retained` directly — if the +unlink that follows inside `writeResult()` also fails, that is caught by the +one `retained` path described above, same as for any other trigger. + +`dispatch()` still wraps the call to `processTriggerFile()` in a `.catch()` +that logs and marks the name `retained`. With the internal `try`/`catch` now +covering the whole function body, this outer `.catch()` should never fire in +practice — a broken `ctx.log` alone can no longer reach it, since every log +call between it and `processTriggerFile()`'s own generic catch is now +`safeLogError()`-guarded too. It stays as a last-resort backstop for the one +thing genuinely outside this module's control: `retained.add(filename)` +(a plain `Set`) itself throwing. `retained.add(filename)` runs *before* the +logging in this `.catch()`, same ordering rule as everywhere else. + +Only two `ctx.log.error()` calls sit downstream of every other one in this +file, and both are `safeLogError()`-guarded: the generic catch's own log line +above, and `dispatch()`'s `.catch()` here. Every other `ctx.log.warn` / +`.info` / `.error()` call inside `processTriggerFile()`'s body is deliberately +*not* wrapped individually — each one precedes a `return` through +`writeResult()` inside the same outer `try`, so a throw from any of them is +already caught by the generic catch above, and (if that catch's own guarded +log and `writeResult()` retry somehow both fail) by `dispatch()`'s catch in +turn. Wrapping every site individually would duplicate that protection +without closing any gap the two backstops don't already close. The four +`ctx.log.*` calls in `start()` itself (directory creation, watcher startup, +the `fs.watch` `error` event) are a different case: they describe the +watcher's own lifecycle, not any one trigger's fate, and are out of scope for +this guarantee. + +**Known gap, deliberately not fixed**: if writing the result file fails, the +trigger is deleted anyway. The two invariants ("always a result", "never twice") +cannot both hold there, and "never twice" wins. + +`processed/` **has no retention policy** — result files accumulate without bound +and nothing prunes them. + ## Invariants -- **Never throws out of the watcher callback** — every error path lands in the result file. +- **Never throws out of the watcher callback** — `processTriggerFile()`'s body runs inside one `try`/`catch`; any exception, anticipated or not, lands in the result file via `writeResult()` before the function returns. `dispatch()`'s own `.catch()` is a backstop for the case that should no longer occur. +- **Poll loops must reject, not throw** — `waitForComposerFree`, `pollForBusyRise`, `waitForBusyFall` and `waitForIdle` all share `pollLoop()`, which converts a throw from *any* tick (including the ones run from inside `setTimeout`, not just the first synchronous one) into that promise's rejection. Without this, a throw on a deferred tick has no `try`/`catch` above it — see "Removing the entry". +- **`writeResult()` never throws** — both of its internal `try`/`catch` blocks route their own logging through `safeLogError()`, which cannot itself throw, and the unlink branch records `onEntryRetained()` before logging. A broken `ctx.log` cannot skip either guarantee. +- **A broken `ctx.log` cannot crash the process from either backstop** — the generic catch's own log line and `dispatch()`'s own `.catch()` log line are both `safeLogError()`-guarded. An unguarded log call throwing there would otherwise become an `unhandledRejection`, which terminates the process by default under Node — see "Removing the entry". - **Deduplication via `inFlight` Set** — noisy `rename` events for the same file (common on Linux inotify) are coalesced; a file is processed at most once per appearance. +- **A processed trigger never runs twice** — normally because it was deleted; when the deletion fails, because its name is in `retained`. A trigger that threw internally is not exempt from this: it still gets a result and a deletion, so it is not "retained" on that account. - **`accessSync` guard** — the `rename` event fires both on file creation and deletion; the existence check prevents processing a deletion event. - **Directories ignored** — non-`*.json` filenames and any name containing `/` or `path.sep` are skipped. - **Invalid `timeout_ms` releases the semaphore** — validation happens before the session look-up and before acquiring an idle-wait slot; a bad value produces a result file and returns without counting against `MAX_INFLIGHT`. diff --git a/docs/automation.md b/docs/automation.md index 36db1b55..6b79cdee 100644 --- a/docs/automation.md +++ b/docs/automation.md @@ -205,7 +205,11 @@ away mid-sentence can stall the queue for every other session. Set a short ### Reading a result -The trigger file is deleted after processing, and a result file is written to `~/.switchboard/triggers/processed/.result.json`: +Every path that decides a trigger's fate — success, validation refusal, timeout, +missing session, refused `wait` — writes the result to +`~/.switchboard/triggers/processed/.result.json` and then deletes the +trigger file. The directory therefore holds exactly the triggers still waiting +to be processed: ```json { "ok": true, "submitted": "confirmed", "sessionId": "...", "command": "...", "sent_at": "...", "waited_ms": 320 } @@ -249,4 +253,29 @@ The two reserved values are easy to confuse and mean opposite things, so: stays the free-text `session exited during wait`, but `submitted` is `no`, `partial` is `false`, and `reason` says nothing was written. +**An exception anywhere while deciding a trigger's fate** — not just the +anticipated validation refusals above — still ends in a result file and a +deletion. A trigger body that parses as valid JSON but isn't a usable shape +(the bare value `null`, a `chain` step that isn't an object) is caught and +reported as `{ "ok": false, "error": "internal error: ", "internal": +true }`, rather than left on disk with no result at all. `internal: true` is +set on this path only — a validation refusal never carries it — so a reader +can tell "our code broke" apart from "the trigger was refused" without +parsing `error`, which stays reserved for the strict-equality checks above. + +**When the deletion itself fails** (permissions, a locked file, an entry that is +not a regular file), the trigger stays on disk. The result file is still +written, the failure is logged at error level, and that name is remembered for +the lifetime of the process so a later filesystem event on it can never run the +command a second time — the leftover file is inert, not pending. A trigger whose +name sits in `processed/` has been processed, whatever the trigger directory +still shows. This is the only case that leaves a name non-replayable; an +internal exception on its own does not — once the result is written and the +trigger deleted, a later trigger dropped under the same name is a fresh +attempt. + +**`processed/` has no retention policy**: result files accumulate there for as +long as the directory lives, and nothing in the app ever removes them. Callers +that write many triggers should prune it themselves. + The primary use case is context-management harnesses — e.g. an agent hook that detects a full context window and injects `/compact` into its own session. Write the trigger file atomically (write to a temp name, then rename) so the watcher never reads a half-written file. diff --git a/test/trigger-watcher.test.js b/test/trigger-watcher.test.js index a262e3d1..801471d0 100644 --- a/test/trigger-watcher.test.js +++ b/test/trigger-watcher.test.js @@ -2470,3 +2470,772 @@ test('renouncing: a chain whose first step was written reports "chain timeout", cleanup(tmp); } }); + +// ── Entry removal after processing ──────────────────────────────────────────── +// See .ai/contexts/trigger-watcher.md — a processed trigger must leave the +// directory, and one that cannot be removed must never be run a second time. + +/** Logger that records what it was told, so failures can be asserted on. */ +function recordingLog() { + const errors = []; + return { + info: () => {}, + warn: () => {}, + debug: () => {}, + error: (...args) => { errors.push(args.join(' ')); }, + _errors: errors, + }; +} + +test('unremovable entry: failure is logged instead of swallowed', async () => { + const tmp = mkTmp(); + let watcher; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-unremovable-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + ctx.log = recordingLog(); + watcher = start(ctx); + + // A directory named .json: lstat succeeds, isFile() is false, so the + // watcher writes a result — and unlink() on a directory always fails. + const uuid = 'unremovable-' + Date.now(); + const entry = path.join(tmp, uuid + '.json'); + fs.mkdirSync(entry); + + const resultPath = path.join(tmp, 'processed', uuid + '.result.json'); + await waitForFile(resultPath); + + const result = readResult(path.join(tmp, 'processed'), uuid); + assert.equal(result.ok, false); + assert.match(result.error, /regular file/); + assert.equal(fs.existsSync(entry), true, 'precondition: the entry cannot be unlinked'); + + assert.ok( + ctx.log._errors.some(m => /survived processing/.test(m)), + 'a removal failure must be logged, not swallowed by a bare catch; got: ' + + JSON.stringify(ctx.log._errors), + ); + + } finally { + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('unremovable entry: a later event on the same name is never processed again', async () => { + const tmp = mkTmp(); + let watcher; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-noreplay-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + ctx.log = recordingLog(); + watcher = start(ctx); + + const uuid = 'noreplay-' + Date.now(); + const entry = path.join(tmp, uuid + '.json'); + fs.mkdirSync(entry); + + const processedDir = path.join(tmp, 'processed'); + const resultPath = path.join(processedDir, uuid + '.result.json'); + await waitForFile(resultPath); + assert.equal(readResult(processedDir, uuid).ok, false, 'first pass rejected the entry'); + + // The entry survived processing. Make the same name appear again — a valid + // trigger this time. It must NOT be picked up: it was already processed. + fs.rmdirSync(entry); + writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact' }); + + await new Promise(r => setTimeout(r, 600)); + + assert.deepEqual(ctx._written, [], + 'a name whose entry survived processing must never reach the PTY again'); + assert.equal(readResult(processedDir, uuid).ok, false, + 'the original result must not be overwritten by a second run'); + + } finally { + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('a throwing ctx yields a definitive result (no unhandled rejection), and a later attempt is not blocked', async () => { + const tmp = mkTmp(); + let watcher; + const rejections = []; + const onRejection = (err) => rejections.push(err); + process.on('unhandledRejection', onRejection); + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-throwing-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + ctx.log = recordingLog(); + let calls = 0; + let shouldThrow = true; + ctx.getComposerState = (id) => { + calls++; + if (shouldThrow) throw new Error('composer state unavailable'); + return { pending: 0, lastInputAt: 0 }; + }; + watcher = start(ctx); + + const uuid = 'throwing-' + Date.now(); + const triggerPath = writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact' }); + const processedDir = path.join(tmp, 'processed'); + const resultPath = path.join(processedDir, uuid + '.result.json'); + + await waitForFile(resultPath); + assert.ok(calls > 0, 'precondition: the throwing hook was reached'); + + const firstResult = readResult(processedDir, uuid); + assert.equal(firstResult.ok, false); + assert.match(firstResult.error, /composer state unavailable/, + 'the caught exception surfaces in the result, not just the log'); + assert.equal(fs.existsSync(triggerPath), false, + 'the trigger file must be deleted even though processing threw'); + + assert.deepEqual(rejections, [], 'the watcher must not leave an unhandled rejection'); + assert.ok( + ctx.log._errors.some(m => /processing threw/.test(m)), + 'the failure must be logged; got: ' + JSON.stringify(ctx.log._errors), + ); + + // Nothing was left unresolved by the first attempt — a result was written + // and the trigger was deleted — so a fresh trigger dropped under the same + // name afterwards is a new attempt, not a replay, and must go through. + shouldThrow = false; + fs.rmSync(resultPath); + writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact' }); + await waitForFile(resultPath, 2000); + + const secondResult = readResult(processedDir, uuid); + assert.equal(secondResult.ok, true, 'a fresh trigger with the same name must not be blocked by retained'); + assert.ok(ctx._written.includes('/compact'), 'the second, valid attempt reaches the PTY'); + + } finally { + process.removeListener('unhandledRejection', onRejection); + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +// ── Defects found in adversarial review of PR #166 ──────────────────────────── +// Two paths could decide a trigger's fate without ever calling writeResult(): +// a throw before/during shape validation (destructuring `null`, or a chain +// step that isn't an object), and a non-ENOENT lstat failure. Both used to +// leave the trigger on disk forever with no result file. See +// .ai/contexts/trigger-watcher.md, "Removing the entry". + +test('trigger body is JSON null: destructuring throws, but the entry is still resolved', async () => { + const tmp = mkTmp(); + let watcher; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const ctx = makeCtx('any-session'); + watcher = start(ctx); + + const uuid = 'null-body-' + Date.now(); + const triggerPath = path.join(tmp, uuid + '.json'); + fs.writeFileSync(triggerPath, 'null', 'utf8'); // valid JSON; destructuring it throws + + const resultPath = path.join(tmp, 'processed', uuid + '.result.json'); + await waitForFile(resultPath, 2000); + + const result = readResult(path.join(tmp, 'processed'), uuid); + assert.equal(result.ok, false); + assert.match(result.error, /internal error/i); + assert.equal(fs.existsSync(triggerPath), false, + 'trigger file must be deleted, not left behind forever'); + assert.deepEqual(ctx._written, [], 'no PTY write for a null trigger body'); + + } finally { + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('chain step is not an object: property access throws, but the entry is still resolved', async () => { + const tmp = mkTmp(); + let watcher; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const ctx = makeCtx('any-session'); + watcher = start(ctx); + + const uuid = 'chain-null-step-' + Date.now(); + const triggerPath = writeTrigger(tmp, uuid, { sessionId: 'any-session', chain: [null] }); + + const resultPath = path.join(tmp, 'processed', uuid + '.result.json'); + await waitForFile(resultPath, 2000); + + const result = readResult(path.join(tmp, 'processed'), uuid); + assert.equal(result.ok, false); + assert.match(result.error, /internal error/i); + assert.equal(fs.existsSync(triggerPath), false, + 'trigger file must be deleted, not left behind forever'); + assert.deepEqual(ctx._written, [], 'no PTY write for a chain with a non-object step'); + + } finally { + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('lstat fails with a non-ENOENT error: result written and trigger deleted, not silently returned', async () => { + const tmp = mkTmp(); + let watcher; + const realLstatSync = fs.lstatSync; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-lstat-eperm-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + watcher = start(ctx); + + const uuid = 'lstat-eperm-' + Date.now(); + const triggerPath = path.join(tmp, uuid + '.json'); + + // Simulate a share-lock / permission error a real filesystem can raise — + // distinct from ENOENT, which is the one case this function must still + // treat as "nothing to report" (see the ENOENT branch just above). + fs.lstatSync = (p, ...rest) => { + if (p === triggerPath) { + const err = new Error('EPERM: operation not permitted, lstat ' + p); + err.code = 'EPERM'; + throw err; + } + return realLstatSync.call(fs, p, ...rest); + }; + + writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact' }); + + const resultPath = path.join(tmp, 'processed', uuid + '.result.json'); + await waitForFile(resultPath, 2000); + + const result = readResult(path.join(tmp, 'processed'), uuid); + assert.equal(result.ok, false); + assert.match(result.error, /could not be inspected/i); + assert.equal(fs.existsSync(triggerPath), false, + 'trigger file must be deleted even when lstat itself fails'); + assert.deepEqual(ctx._written, [], 'no PTY write when lstat fails'); + + } finally { + fs.lstatSync = realLstatSync; + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +// ── Defects found in the third adversarial review of PR #166 ────────────────── +// All four poll loops call ctx back on a deferred `setTimeout` tick, not just +// their first, synchronous call. A throw from that deferred tick used to have +// nothing to catch it — not the Promise executor (already returned), not +// processTriggerFile's try/catch, not dispatch()'s .catch(). See +// .ai/contexts/trigger-watcher.md, "Poll loops must reject, not throw". + +test('a hook that throws starting from the SECOND tick of the composer-free poll does not escape as an uncaughtException', async () => { + const tmp = mkTmp(); + let watcher; + const uncaughtErrors = []; + const onUncaught = (err) => uncaughtErrors.push(err); + process.on('uncaughtException', onUncaught); + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '2000'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-composer-deferred-throw-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + ctx.log = recordingLog(); + let calls = 0; + ctx.getComposerState = (id) => { + calls++; + if (calls === 1) { + // Not free -> forces the setTimeout-based recheck, never resolved + // from inside the Promise executor's synchronous frame again. + return { pending: 5, lastInputAt: Date.now() }; + } + throw new Error('composer state unavailable (deferred)'); + }; + watcher = start(ctx); + + const uuid = 'composer-deferred-throw-' + Date.now(); + const triggerPath = writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact' }); + const processedDir = path.join(tmp, 'processed'); + const resultPath = path.join(processedDir, uuid + '.result.json'); + + await waitForFile(resultPath, 2000); + + assert.ok(calls >= 2, + 'precondition: the throw happened on a deferred tick, not the first synchronous call'); + assert.deepEqual(uncaughtErrors, [], + 'a throw on a deferred poll tick must not become an uncaughtException'); + + const result = readResult(processedDir, uuid); + assert.equal(result.ok, false); + assert.match(result.error, /composer state unavailable \(deferred\)/); + assert.equal(fs.existsSync(triggerPath), false, + 'trigger file must be deleted even though the throw happened on a deferred tick'); + + } finally { + process.removeListener('uncaughtException', onUncaught); + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('a hook that throws starting from the SECOND tick of the idle-wait poll does not escape as an uncaughtException', async () => { + const tmp = mkTmp(); + let watcher; + const uncaughtErrors = []; + const onUncaught = (err) => uncaughtErrors.push(err); + process.on('uncaughtException', onUncaught); + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '2000'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-idle-deferred-throw-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + ctx.log = recordingLog(); + let calls = 0; + ctx.isSessionBusy = (id) => { + calls++; + if (calls === 1) return true; // busy -> forces the setTimeout-based recheck + throw new Error('busy check unavailable (deferred)'); + }; + watcher = start(ctx); + + const uuid = 'idle-deferred-throw-' + Date.now(); + const triggerPath = writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact', wait: 'idle' }); + const processedDir = path.join(tmp, 'processed'); + const resultPath = path.join(processedDir, uuid + '.result.json'); + + await waitForFile(resultPath, 2000); + + assert.ok(calls >= 2, + 'precondition: the throw happened on a deferred tick, not the first synchronous call'); + assert.deepEqual(uncaughtErrors, [], + 'a throw on a deferred idle-wait tick must not become an uncaughtException'); + + const result = readResult(processedDir, uuid); + assert.equal(result.ok, false); + assert.match(result.error, /busy check unavailable \(deferred\)/); + assert.equal(fs.existsSync(triggerPath), false, + 'trigger file must be deleted even though the throw happened on a deferred tick'); + + } finally { + process.removeListener('uncaughtException', onUncaught); + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('writeResult never throws even when ctx.log.error itself throws (no uncaughtException, no unhandledRejection)', async () => { + const tmp = mkTmp(); + let watcher; + const uncaughtErrors = []; + const rejections = []; + const onUncaught = (err) => uncaughtErrors.push(err); + const onRejection = (err) => rejections.push(err); + process.on('uncaughtException', onUncaught); + process.on('unhandledRejection', onRejection); + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-log-throws-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + let logCalls = 0; + ctx.log = { + info: () => {}, warn: () => {}, debug: () => {}, + error: (...a) => { logCalls++; throw new Error('logger is broken'); }, + }; + watcher = start(ctx); + + // A directory named .json: lstat succeeds, isFile() is false -> + // writeResult({ok:false}) runs; unlink() on a directory then fails + // (non-ENOENT), reaching the retained path whose own log call throws. + const uuid = 'log-throws-' + Date.now(); + const entry = path.join(tmp, uuid + '.json'); + fs.mkdirSync(entry); + + const processedDir = path.join(tmp, 'processed'); + const resultPath = path.join(processedDir, uuid + '.result.json'); + + await waitForFile(resultPath, 2000); + + assert.ok(logCalls > 0, 'precondition: the throwing logger was reached'); + assert.deepEqual(uncaughtErrors, [], 'a broken logger must not surface as an uncaughtException'); + assert.deepEqual(rejections, [], 'a broken logger must not surface as an unhandledRejection'); + + const result = readResult(processedDir, uuid); + assert.equal(result.ok, false); + assert.match(result.error, /regular file/); + assert.equal(fs.existsSync(entry), true, + 'the entry could not be unlinked and must stay on disk (retained)'); + + } finally { + process.removeListener('uncaughtException', onUncaught); + process.removeListener('unhandledRejection', onRejection); + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('benign ENOENT race on unlink does not retain the name: a later reuse of the same uuid is processed', async () => { + const tmp = mkTmp(); + let watcher; + const realUnlinkSync = fs.unlinkSync; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '2000'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-enoent-race-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + ctx.log = recordingLog(); + watcher = start(ctx); + + const uuid = 'enoent-race-' + Date.now(); + const triggerPath = path.join(tmp, uuid + '.json'); + const processedDir = path.join(tmp, 'processed'); + const resultPath = path.join(processedDir, uuid + '.result.json'); + + // Fake an external actor deleting the trigger file behind our back, just + // before our own unlinkSync call — the documented "benign ENOENT" race + // between two events on the same file (see .ai/contexts/trigger-watcher.md, + // "Removing the entry"). + let sawUnlinkAttempt = false; + fs.unlinkSync = (p, ...rest) => { + if (p === triggerPath && !sawUnlinkAttempt) { + sawUnlinkAttempt = true; + try { realUnlinkSync.call(fs, p); } catch {} + const err = new Error('ENOENT: no such file or directory, unlink ' + p); + err.code = 'ENOENT'; + throw err; + } + return realUnlinkSync.call(fs, p, ...rest); + }; + + writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact' }); + await waitForFile(resultPath, 2000); + assert.equal(readResult(processedDir, uuid).ok, true, 'first pass processed normally'); + + fs.unlinkSync = realUnlinkSync; + assert.ok( + !ctx.log._errors.some(m => /survived processing/.test(m)), + 'ENOENT on unlink must stay silent, not be logged as a survived entry; got: ' + + JSON.stringify(ctx.log._errors), + ); + + // A brand-new, legitimate trigger reuses the same uuid (e.g. a retried + // harness call). It must be picked up like any fresh trigger, not ignored + // as if the name had been retained. + fs.rmSync(resultPath); + writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact-second' }); + + await waitForFile(resultPath, 2000); + assert.equal(readResult(processedDir, uuid).ok, true, + 'a name freed by a benign ENOENT race must not stay retained'); + assert.ok(ctx._written.includes('/compact-second'), + 'the reused trigger must reach the PTY, not be silently ignored'); + + } finally { + fs.unlinkSync = realUnlinkSync; + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('writeResult never throws when the result write itself fails AND ctx.log.error throws (both branches guarded)', async () => { + const tmp = mkTmp(); + let watcher; + const uncaughtErrors = []; + const rejections = []; + const onUncaught = (err) => uncaughtErrors.push(err); + const onRejection = (err) => rejections.push(err); + process.on('uncaughtException', onUncaught); + process.on('unhandledRejection', onRejection); + const realWriteFileSync = fs.writeFileSync; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const SESSION_ID = 'sess-write-fails-' + Date.now(); + const ctx = makeCtx(SESSION_ID); + let logCalls = 0; + ctx.log = { + info: () => {}, warn: () => {}, debug: () => {}, + error: (...a) => { logCalls++; throw new Error('logger is broken'); }, + }; + watcher = start(ctx); + + const uuid = 'write-fails-' + Date.now(); + const triggerPath = writeTrigger(tmp, uuid, { sessionId: SESSION_ID, command: '/compact' }); + const processedDir = path.join(tmp, 'processed'); + const resultTmpPath = path.join(processedDir, uuid + '.result.json.tmp'); + + // Force the .tmp write inside writeResult() to fail, so its own catch's + // (now-guarded) log call is exercised — the branch the previous mutation + // probe found untested. + fs.writeFileSync = (p, ...rest) => { + if (p === resultTmpPath) throw new Error('disk full (simulated)'); + return realWriteFileSync.call(fs, p, ...rest); + }; + + // Poll for the trigger file being gone rather than for a result file — + // the result write is the thing we are forcing to fail. + const deadline = Date.now() + 2000; + while (fs.existsSync(triggerPath) && Date.now() < deadline) { + await new Promise(r => setTimeout(r, 20)); + } + + assert.ok(logCalls > 0, 'precondition: the throwing logger was reached'); + assert.deepEqual(uncaughtErrors, [], 'a broken logger must not surface as an uncaughtException'); + assert.deepEqual(rejections, [], 'a broken logger must not surface as an unhandledRejection'); + assert.equal(fs.existsSync(triggerPath), false, + 'the unlink must still run even though the result write failed first'); + + } finally { + fs.writeFileSync = realWriteFileSync; + process.removeListener('uncaughtException', onUncaught); + process.removeListener('unhandledRejection', onRejection); + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('internal field: the generic catch marks internal:true; a validation refusal does not', async () => { + const tmp = mkTmp(); + let watcher; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const ctx = makeCtx('any-session'); + watcher = start(ctx); + + // A trigger body that parses as JSON but destructures wrong -> caught by + // the generic catch at the end of processTriggerFile. + const uuidInternal = 'internal-flag-' + Date.now(); + const triggerPathInternal = path.join(tmp, uuidInternal + '.json'); + fs.writeFileSync(triggerPathInternal, 'null', 'utf8'); + const resultPathInternal = path.join(tmp, 'processed', uuidInternal + '.result.json'); + await waitForFile(resultPathInternal, 2000); + const internalResult = readResult(path.join(tmp, 'processed'), uuidInternal); + assert.equal(internalResult.internal, true, + 'a generic caught exception must be marked internal:true, distinguishable from a refusal'); + + // A plain validation refusal must NOT carry internal:true. + const uuidRefusal = 'refusal-flag-' + Date.now(); + writeTrigger(tmp, uuidRefusal, { sessionId: '' }); // missing required field: sessionId + const resultPathRefusal = path.join(tmp, 'processed', uuidRefusal + '.result.json'); + await waitForFile(resultPathRefusal, 2000); + const refusalResult = readResult(path.join(tmp, 'processed'), uuidRefusal); + assert.equal(refusalResult.internal, undefined, + 'a validation refusal must not be indistinguishable from an internal bug'); + + } finally { + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +// ── Follow-up: the two remaining unguarded ctx.log.error() call sites ───────── +// (raised after the first pass at these fixes). Both sit downstream of every +// other log call in this file — anything that throws upstream is caught by +// the outer try/catch in processTriggerFile() and lands here, or (if that +// itself throws) in dispatch()'s .catch(). A broken ctx.log at either site +// used to end in an unhandledRejection, which terminates the process by +// default under Node. See .ai/contexts/trigger-watcher.md, "Removing the +// entry". + +test('outer generic-catch log call cannot escape even when ctx.log.error throws (result still written, trigger still deleted)', async () => { + const tmp = mkTmp(); + let watcher; + const uncaughtErrors = []; + const rejections = []; + const onUncaught = (err) => uncaughtErrors.push(err); + const onRejection = (err) => rejections.push(err); + process.on('uncaughtException', onUncaught); + process.on('unhandledRejection', onRejection); + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const ctx = makeCtx('any-session'); + let logCalls = 0; + ctx.log = { + info: () => {}, warn: () => {}, debug: () => {}, + error: (...a) => { logCalls++; throw new Error('logger is broken'); }, + }; + watcher = start(ctx); + + // A trigger body that parses as JSON but destructures wrong -> reaches + // the generic catch at the end of processTriggerFile, whose own log call + // is the site under test. + const uuid = 'outer-catch-log-throws-' + Date.now(); + const triggerPath = path.join(tmp, uuid + '.json'); + fs.writeFileSync(triggerPath, 'null', 'utf8'); + + const processedDir = path.join(tmp, 'processed'); + const resultPath = path.join(processedDir, uuid + '.result.json'); + await waitForFile(resultPath, 2000); + + assert.ok(logCalls > 0, 'precondition: the throwing logger was reached'); + assert.deepEqual(uncaughtErrors, [], + 'a broken logger in the outer catch must not surface as an uncaughtException'); + assert.deepEqual(rejections, [], + 'a broken logger in the outer catch must not surface as an unhandledRejection'); + + const result = readResult(processedDir, uuid); + assert.equal(result.ok, false); + assert.equal(result.internal, true); + assert.equal(fs.existsSync(triggerPath), false, + 'trigger must still be deleted despite the broken logger'); + + } finally { + process.removeListener('uncaughtException', onUncaught); + process.removeListener('unhandledRejection', onRejection); + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); + +test('dispatch() backstop log call cannot escape even when ctx.log.error throws (name still ends up retained)', async () => { + const tmp = mkTmp(); + let watcher; + const uncaughtErrors = []; + const rejections = []; + const onUncaught = (err) => uncaughtErrors.push(err); + const onRejection = (err) => rejections.push(err); + process.on('uncaughtException', onUncaught); + process.on('unhandledRejection', onRejection); + + // dispatch()'s own .catch() only fires if processTriggerFile() itself + // rejects -- which, with both writeResult() try/catch blocks now safe, + // only still happens if onEntryRetained() (== retained.add(filename), a + // Set the caller never sees) throws. This forces exactly that, to reach + // the one remaining call site without touching module internals: a + // directory-shaped trigger makes writeResult()'s unlink fail twice (the + // validation-refusal write, then the outer catch's own fallback write), + // so Set.prototype.add is patched to throw only for the 2nd and 3rd + // .add() call carrying this trigger's exact filename -- letting + // dispatch()'s own (4th) retained.add(filename) call go through for real, + // which is the guarantee under test. + const realSetAdd = Set.prototype.add; + let addCallsForFile = 0; + try { + process.env.SWITCHBOARD_TRIGGERS_DIR = tmp; + process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS = '200'; + + const { start } = require('../trigger-watcher'); + const ctx = makeCtx('any-session'); + let logCalls = 0; + ctx.log = { + info: () => {}, warn: () => {}, debug: () => {}, + error: (...a) => { logCalls++; throw new Error('logger is broken'); }, + }; + watcher = start(ctx); + + const uuid = 'dispatch-catch-log-throws-' + Date.now(); + const entry = path.join(tmp, uuid + '.json'); // directory: unlink always fails + const filename = uuid + '.json'; + + Set.prototype.add = function (value) { + if (value === filename) { + addCallsForFile++; + if (addCallsForFile === 2 || addCallsForFile === 3) { + throw new Error('retained set is broken (simulated)'); + } + } + return realSetAdd.call(this, value); + }; + + fs.mkdirSync(entry); + + // No result-file poll: the write inside writeResult() races the induced + // throw, so poll for the trigger to stop being reprocessed instead. + const deadline = Date.now() + 2000; + while (addCallsForFile < 4 && Date.now() < deadline) { + await new Promise(r => setTimeout(r, 20)); + } + await new Promise(r => setTimeout(r, 100)); // let dispatch()'s .finally() settle + + Set.prototype.add = realSetAdd; + + assert.ok(addCallsForFile >= 4, + 'precondition: retained.add(filename) was reached a 4th time, from dispatch()\'s own catch'); + assert.ok(logCalls > 0, 'precondition: the throwing logger was reached'); + assert.deepEqual(uncaughtErrors, [], + 'a broken logger in dispatch()\'s backstop must not surface as an uncaughtException'); + assert.deepEqual(rejections, [], + 'a broken logger in dispatch()\'s backstop must not surface as an unhandledRejection'); + + // The name must be genuinely retained: drop a fresh, valid trigger under + // the same uuid and confirm it is never picked up. + fs.rmdirSync(entry); + writeTrigger(tmp, uuid, { sessionId: 'any-session', command: '/compact' }); + await new Promise(r => setTimeout(r, 400)); + assert.deepEqual(ctx._written, [], + 'the name must stay retained -- dispatch()\'s own retained.add(filename) must have gone through'); + + } finally { + Set.prototype.add = realSetAdd; + process.removeListener('uncaughtException', onUncaught); + process.removeListener('unhandledRejection', onRejection); + if (watcher) watcher.close(); + delete process.env.SWITCHBOARD_TRIGGERS_DIR; + delete process.env.SWITCHBOARD_TRIGGER_IDLE_TIMEOUT_MS; + cleanup(tmp); + } +}); diff --git a/trigger-watcher.js b/trigger-watcher.js index 42473284..e6ac89eb 100644 --- a/trigger-watcher.js +++ b/trigger-watcher.js @@ -132,6 +132,26 @@ function describeBusyComposer(state, quietMs, now) { `inside the ${quietMs} ms quiet window`; } +/** + * Runs `check(resolve, scheduleNext)`, re-invoking it on every recursive + * `scheduleNext()`; any throw from `check`, on any tick, becomes this + * promise's rejection. See .ai/contexts/trigger-watcher.md, "Poll loops must + * reject, not throw". + */ +function pollLoop(check) { + return new Promise((resolve, reject) => { + function tick() { + try { + // .unref(): an in-flight poll must not keep the process alive alone. + check(resolve, () => setTimeout(tick, IDLE_POLL_INTERVAL).unref()); + } catch (err) { + reject(err); + } + } + tick(); + }); +} + /** * Poll until the target composer is free — empty AND quiet — bounded by the * absolute `deadlineMs`. @@ -142,31 +162,27 @@ function describeBusyComposer(state, quietMs, now) { * Returns { free, waited_ms, reason }. */ function waitForComposerFree(sessionId, ctx, deadlineMs) { - return new Promise((resolve) => { - const start = Date.now(); - const quietMs = getQuietMs(); - const limit = (deadlineMs !== undefined) ? deadlineMs : Infinity; - - function check() { - const now = Date.now(); - const state = (typeof ctx.getComposerState === 'function') - ? ctx.getComposerState(sessionId) - : null; - - if (state && state.pending === 0 && (now - (state.lastInputAt || 0)) >= quietMs) { - return resolve({ free: true, waited_ms: now - start, reason: null }); - } - if (now >= limit) { - return resolve({ - free: false, - waited_ms: now - start, - reason: describeBusyComposer(state, quietMs, now), - }); - } - setTimeout(check, IDLE_POLL_INTERVAL).unref(); + const start = Date.now(); + const quietMs = getQuietMs(); + const limit = (deadlineMs !== undefined) ? deadlineMs : Infinity; + + return pollLoop((resolve, scheduleNext) => { + const now = Date.now(); + const state = (typeof ctx.getComposerState === 'function') + ? ctx.getComposerState(sessionId) + : null; + + if (state && state.pending === 0 && (now - (state.lastInputAt || 0)) >= quietMs) { + return resolve({ free: true, waited_ms: now - start, reason: null }); } - - check(); + if (now >= limit) { + return resolve({ + free: false, + waited_ms: now - start, + reason: describeBusyComposer(state, quietMs, now), + }); + } + scheduleNext(); }); } @@ -188,33 +204,26 @@ function getSubmitVerifyMs() { * - sessionExited: PTY vanished during the poll */ function pollForBusyRise(sessionId, ctx, windowMs, deadlineMs) { - return new Promise((resolve) => { - const start = Date.now(); - const windowEnd = start + windowMs; + const start = Date.now(); + const windowEnd = start + windowMs; - function check() { - const now = Date.now(); + return pollLoop((resolve, scheduleNext) => { + const now = Date.now(); - if (now >= deadlineMs) { - return resolve({ rose: false, timedOut: true, sessionExited: false, waited_ms: now - start }); - } - if (!ctx.getPtyForSession(sessionId)) { - return resolve({ rose: false, timedOut: false, sessionExited: true, waited_ms: now - start }); - } - if (ctx.isSessionBusy(sessionId)) { - return resolve({ rose: true, timedOut: false, sessionExited: false, waited_ms: now - start }); - } - if (now >= windowEnd) { - // Verify window elapsed without a rise — caller decides what to do. - return resolve({ rose: false, timedOut: false, sessionExited: false, waited_ms: now - start }); - } - // .unref() so an in-flight poll never keeps the process alive on its own. - // In the app, other handles keep the loop running; in tests this lets the - // runner exit cleanly instead of hanging until DEFAULT_IDLE_TIMEOUT (5 min). - setTimeout(check, IDLE_POLL_INTERVAL).unref(); + if (now >= deadlineMs) { + return resolve({ rose: false, timedOut: true, sessionExited: false, waited_ms: now - start }); } - - check(); + if (!ctx.getPtyForSession(sessionId)) { + return resolve({ rose: false, timedOut: false, sessionExited: true, waited_ms: now - start }); + } + if (ctx.isSessionBusy(sessionId)) { + return resolve({ rose: true, timedOut: false, sessionExited: false, waited_ms: now - start }); + } + if (now >= windowEnd) { + // Verify window elapsed without a rise — caller decides what to do. + return resolve({ rose: false, timedOut: false, sessionExited: false, waited_ms: now - start }); + } + scheduleNext(); }); } @@ -306,27 +315,20 @@ async function submitWithVerify(ptyProcess, sessionId, command, ctx, deadlineMs) * Returns { timedOut, sessionExited, waited_ms }. */ function waitForBusyFall(sessionId, ctx, deadlineMs) { - return new Promise((resolve) => { - const start = Date.now(); + const start = Date.now(); - function check() { - const now = Date.now(); - if (now >= deadlineMs) { - return resolve({ timedOut: true, sessionExited: false, waited_ms: now - start }); - } - if (!ctx.getPtyForSession(sessionId)) { - return resolve({ timedOut: false, sessionExited: true, waited_ms: now - start }); - } - if (!ctx.isSessionBusy(sessionId)) { - return resolve({ timedOut: false, sessionExited: false, waited_ms: now - start }); - } - // .unref() so an in-flight poll never keeps the process alive on its own. - // In the app, other handles keep the loop running; in tests this lets the - // runner exit cleanly instead of hanging until DEFAULT_IDLE_TIMEOUT (5 min). - setTimeout(check, IDLE_POLL_INTERVAL).unref(); + return pollLoop((resolve, scheduleNext) => { + const now = Date.now(); + if (now >= deadlineMs) { + return resolve({ timedOut: true, sessionExited: false, waited_ms: now - start }); } - - check(); + if (!ctx.getPtyForSession(sessionId)) { + return resolve({ timedOut: false, sessionExited: true, waited_ms: now - start }); + } + if (!ctx.isSessionBusy(sessionId)) { + return resolve({ timedOut: false, sessionExited: false, waited_ms: now - start }); + } + scheduleNext(); }); } @@ -354,31 +356,24 @@ function getIdleTimeout() { * Returns { timedOut: boolean, sessionExited: boolean, waited_ms: number }. */ function waitForIdle(sessionId, ctx, timeoutMs) { - return new Promise((resolve) => { - const timeout = (timeoutMs !== undefined) ? timeoutMs : getIdleTimeout(); - const start = Date.now(); - - function check() { - const waited_ms = Date.now() - start; + const timeout = (timeoutMs !== undefined) ? timeoutMs : getIdleTimeout(); + const start = Date.now(); - // W5: detect PTY closure during wait - if (!ctx.getPtyForSession(sessionId)) { - return resolve({ timedOut: false, sessionExited: true, waited_ms }); - } + return pollLoop((resolve, scheduleNext) => { + const waited_ms = Date.now() - start; - if (!ctx.isSessionBusy(sessionId)) { - return resolve({ timedOut: false, sessionExited: false, waited_ms }); - } - if (waited_ms >= timeout) { - return resolve({ timedOut: true, sessionExited: false, waited_ms }); - } - // .unref() so an in-flight poll never keeps the process alive on its own. - // In the app, other handles keep the loop running; in tests this lets the - // runner exit cleanly instead of hanging until DEFAULT_IDLE_TIMEOUT (5 min). - setTimeout(check, IDLE_POLL_INTERVAL).unref(); + // W5: detect PTY closure during wait + if (!ctx.getPtyForSession(sessionId)) { + return resolve({ timedOut: false, sessionExited: true, waited_ms }); } - check(); + if (!ctx.isSessionBusy(sessionId)) { + return resolve({ timedOut: false, sessionExited: false, waited_ms }); + } + if (waited_ms >= timeout) { + return resolve({ timedOut: true, sessionExited: false, waited_ms }); + } + scheduleNext(); }); } @@ -406,11 +401,19 @@ function validateTimeoutMs(value) { return null; } +// A logger call that cannot throw. See .ai/contexts/trigger-watcher.md, +// "Removing the entry". +function safeLogError(ctx, ...args) { + try { + ctx.log.error(...args); + } catch (_) { /* swallow */ } +} + /** * Process a single trigger file (by basename, e.g. "abc-123.json"). * Never throws — all errors land in the result file. */ -async function processTriggerFile(name, ctx, triggersDir, processedDir) { +async function processTriggerFile(name, ctx, triggersDir, processedDir, onEntryRetained) { // Only handle *.json files, ignore the processed/ subdir itself and // any stray files. if (!name.endsWith('.json')) return; @@ -429,22 +432,31 @@ async function processTriggerFile(name, ctx, triggersDir, processedDir) { fs.writeFileSync(resultTmp, JSON.stringify(result) + '\n', 'utf8'); fs.renameSync(resultTmp, resultPath); } catch (err) { - ctx.log.error('[trigger-watcher] Failed to write result file:', err.message); + safeLogError(ctx, '[trigger-watcher] Failed to write result file:', err.message); } try { fs.unlinkSync(triggerPath); - } catch { - // Trigger may already be gone (race between two watcher events for the - // same file). Silently ignore. + } catch (err) { + if (err.code !== 'ENOENT') { + // onEntryRetained() first: the guarantee it records must not depend + // on whether the log call after it succeeds. + if (onEntryRetained) onEntryRetained(); + safeLogError(ctx, '[trigger-watcher] Trigger file survived processing, will not be run again:', + name, err.message); + } } } + try { + // ── 1. lstat + size guard (C1 + C2) ────────────────────────────────────── let stat; try { stat = fs.lstatSync(triggerPath); // C2: lstat does NOT follow symlinks - } catch { - // File gone between access check and here — skip silently + } catch (err) { + if (err.code === 'ENOENT') return; // gone before we could inspect it + ctx.log.error('[trigger-watcher] Trigger file could not be inspected:', name, err.message); + await writeResult({ ok: false, error: 'trigger could not be inspected: ' + err.message }); return; } @@ -919,6 +931,13 @@ async function processTriggerFile(name, ctx, triggersDir, processedDir) { steps, total_waited_ms: totalWaitedMs, }); + + } catch (err) { + // see .ai/contexts/trigger-watcher.md, "Removing the entry" + safeLogError(ctx, '[trigger-watcher] Trigger processing threw, writing a generic failure result:', + name, err && err.message); + await writeResult({ ok: false, error: 'internal error: ' + (err && err.message), internal: true }); + } } /** @@ -953,22 +972,31 @@ function start(ctx) { const inFlight = new Set(); // W4: queue of filenames awaiting an in-flight slot const waitQueue = []; + // Entries that were processed but stayed in the directory — see + // .ai/contexts/trigger-watcher.md. + const retained = new Set(); function scheduleNext() { while (waitQueue.length > 0 && inFlight.size < MAX_INFLIGHT) { const filename = waitQueue.shift(); // Dedup: may have been enqueued twice before a slot opened - if (inFlight.has(filename)) continue; + if (inFlight.has(filename) || retained.has(filename)) continue; dispatch(filename); } } function dispatch(filename) { inFlight.add(filename); - processTriggerFile(filename, ctx, triggersDir, processedDir).finally(() => { - inFlight.delete(filename); - scheduleNext(); - }); + processTriggerFile(filename, ctx, triggersDir, processedDir, () => retained.add(filename)) + .catch((err) => { + retained.add(filename); + safeLogError(ctx, '[trigger-watcher] Trigger processing threw, will not be run again:', + filename, err && err.message); + }) + .finally(() => { + inFlight.delete(filename); + scheduleNext(); + }); } let watcher; @@ -980,7 +1008,7 @@ function start(ctx) { // Linux only reports the basename for non-recursive watches, but be // defensive: skip anything that looks like a path separator. if (filename.includes('/') || filename.includes(path.sep)) return; - if (inFlight.has(filename)) return; + if (inFlight.has(filename) || retained.has(filename)) return; // Confirm the file still exists (the rename event fires on delete too) const filePath = path.join(triggersDir, filename);