diff --git a/packages/daemon/src/dispatch.ts b/packages/daemon/src/dispatch.ts index e617d4f..3432356 100644 --- a/packages/daemon/src/dispatch.ts +++ b/packages/daemon/src/dispatch.ts @@ -599,7 +599,6 @@ export interface DispatcherDeps { export class TurnDispatcher { readonly #deps: DispatcherDeps readonly #inFlight = new Map; label: string }>() - readonly #lastReject = new Map() /** Predecessor branches currently spoken for, `branch → the request uri holding it`. Two v2 * requests off ONE predecessor resolve the same branch, and `#inFlight` is keyed by request uri, * so with `concurrency > 1` they would push to it at the same time. Reserved on the CANDIDATE @@ -607,10 +606,8 @@ export class TurnDispatcher { * synchronously to be a reservation at all. Over-serializing a pair whose reuse is later declined * costs one pump interval; interleaving two turns on one branch costs the branch. */ readonly #activeBranches = new Map() - /** Requests skipped for branch contention, so the log says it once rather than every pump. */ + /** Branch contention is exceptional enough to report once while the competing turn runs. */ readonly #contended = new Set() - /** Requests waiting for a serialized harness, logged once rather than on every pump. */ - readonly #serializedContended = new Set() readonly #activeSerializedHarnesses = new Set() constructor(deps: DispatcherDeps) { @@ -656,15 +653,11 @@ export class TurnDispatcher { /** Launches eligible turns from `index` up to remaining concurrency; returns immediately. */ pump(index: MaterializedIndex, actors: ActorRegistry): void { const now = this.#deps.now?.() - const seenRejects = new Set() const heldClaims = this.#deps.heldClaims?.() ?? new Set() - const dispatchable = selectDispatchable(index, actors, this.#deps.ledger, now, (uri, reason) => { - seenRejects.add(uri) - if (this.#lastReject.get(uri) !== reason) { - this.#lastReject.set(uri, reason) - this.#deps.log?.(`not dispatching ${uri}: ${reason}`) - } - }, heldClaims) + // Rejections are ordinary on every sync tick (requests for another operator, exhausted + // capacity, unresolved claims, and so on). Keep them out of the daemon log; actionable turn + // failures are still logged by the launch and settlement paths below. + const dispatchable = selectDispatchable(index, actors, this.#deps.ledger, now, undefined, heldClaims) // The claim view is a cache of the fold, and a cache can be stale in the dangerous direction: // `heldClaims` still naming a request whose claim another operator has since won. So the winner // is re-read from THIS index, immediately before launching, for every unassigned request. @@ -674,14 +667,8 @@ export class TurnDispatcher { winningClaims.set(requestUri, claim.did) } } - for (const uri of this.#lastReject.keys()) if (!seenRejects.has(uri)) this.#lastReject.delete(uri) const selected = new Set(dispatchable.map((item) => item.request.uri)) for (const uri of this.#contended) if (!selected.has(uri)) this.#contended.delete(uri) - if (dispatchable.length > 0) { - this.#deps.log?.( - `${dispatchable.length} dispatchable request(s); ${this.#inFlight.size}/${this.#deps.concurrency} in flight`, - ) - } for (const item of dispatchable) { if (this.#inFlight.size >= this.#deps.concurrency) break @@ -690,9 +677,6 @@ export class TurnDispatcher { // C2: the symmetric guard. `selectDispatchable` already checked this against the same index, // so reaching here means the two disagree — never launch on the strength of the cache alone. if (!item.request.value.assignee && winningClaims.get(uri) !== item.actor.did) { - this.#deps.log?.( - `not dispatching ${uri}: the winning claim is ${winningClaims.get(uri) ?? 'absent'}, not ${item.actor.did}`, - ) continue } if (item.artifactType.name === IMPLEMENTATION_TYPE) { @@ -729,15 +713,8 @@ export class TurnDispatcher { ? item.actor.harness : undefined if (serializedHarness && this.#activeSerializedHarnesses.has(serializedHarness)) { - if (!this.#serializedContended.has(uri)) { - this.#serializedContended.add(uri) - this.#deps.log?.( - `not dispatching ${uri}: harness ${serializedHarness} uses a shared refreshable login; retrying after its active turn finishes`, - ) - } continue } - this.#serializedContended.delete(uri) // Contention, not an error: skip and retry on a later pump, leaving the ledger row untouched // (this runs BEFORE markRunning, so the request stays plainly eligible). diff --git a/packages/daemon/src/turn.ts b/packages/daemon/src/turn.ts index 0f21484..d4a5fa6 100644 --- a/packages/daemon/src/turn.ts +++ b/packages/daemon/src/turn.ts @@ -242,8 +242,8 @@ export function modelEnvSecrets(modelEnv: Record | undefined): s * redaction. Holding each partial line also keeps the daemon log readable when a process writes * one line in several chunks. * - * What a line MEANS is the harness's business (`Harness.renderOutput`); a renderer that declines — - * or a harness with none at all — leaves the raw line, which is what non-JSON CLI output gets too. + * Lines are parsed only for model provenance. Agent activity is intentionally not copied into the + * daemon's operational log; the complete redacted stdout/stderr remains in the per-turn log file. */ function liveOutputLogger( log: ((message: string) => void) | undefined, @@ -261,7 +261,6 @@ function liveOutputLogger( let effectiveModel = initialModel const emit = (stream: ContainerOutputStream, line: string): void => { if (line.length === 0) return - let messages = [line] if (stream === 'stdout' && render) { // Belt and braces: a renderer is contractually no-throw, but it parses untrusted container // output, and a throw here would escape into the runner's output callback mid-turn. @@ -283,12 +282,8 @@ function liveOutputLogger( (previous !== undefined ? ` (switched from ${redactOutput(previous, secrets)})` : ''), ) } - messages = rendered.lines } } - for (const message of messages) { - log?.(`agent ${profile} ${requestUri} ${stream}: ${redactOutput(message, secrets)}`) - } } const write = (stream: ContainerOutputStream, chunk: string): void => { pending[stream] += chunk diff --git a/packages/daemon/test/dispatch.test.mjs b/packages/daemon/test/dispatch.test.mjs index eb04981..0cb5e93 100644 --- a/packages/daemon/test/dispatch.test.mjs +++ b/packages/daemon/test/dispatch.test.mjs @@ -2420,7 +2420,7 @@ it('does not launch overlapping turns for a harness with one refreshable login', dispatcher.pump(index, actors) await new Promise((resolve) => setImmediate(resolve)) assert.equal(started.length, 1) - assert.ok(logs.some((entry) => entry.includes('shared refreshable login')), logs.join(' | ')) + assert.ok(logs.every((entry) => !entry.startsWith('not dispatching ')), logs.join(' | ')) const waiting = [request.uri, second.uri].find((uri) => uri !== started[0]) assert.equal(ledger.get(waiting), undefined) diff --git a/packages/daemon/test/turn.test.mjs b/packages/daemon/test/turn.test.mjs index f39aa15..bd79207 100644 --- a/packages/daemon/test/turn.test.mjs +++ b/packages/daemon/test/turn.test.mjs @@ -358,7 +358,7 @@ it('crashed: a container that produces no socket observation is classified crash }) }) -it('streams tagged, redacted agent output while the turn is running', async () => { +it('keeps streamed agent output out of the daemon log', async () => { const { actor } = await makeActor('did:plc:agent-live-output') await withRunDir(async (runDir) => { const messages = [] @@ -375,15 +375,11 @@ it('streams tagged, redacted agent output while the turn is running', async () = makeToken: () => 'secret-token', log: (message) => messages.push(message), }) - assert.deepEqual(messages.slice(1), [ - `agent planner ${BUNDLE.request.uri} stdout: working with [redacted]`, - `agent planner ${BUNDLE.request.uri} stderr: still running`, - `turn ${BUNDLE.request.uri} exited without an artifact or question; marked crashed`, - ]) + assert.deepEqual(messages.slice(1), [`turn ${BUNDLE.request.uri} exited without an artifact or question; marked crashed`]) }) }) -it('formats Claude stream-json activity for the daemon log', async () => { +it('does not copy formatted Claude activity into the daemon log', async () => { const { actor } = await makeActor('did:plc:agent-stream-json') await withRunDir(async (runDir) => { const messages = [] @@ -404,12 +400,7 @@ it('formats Claude stream-json activity for the daemon log', async () => { checkout: noopCheckout, log: (message) => messages.push(message), }) - assert.deepEqual(messages.slice(1), [ - `agent planner ${BUNDLE.request.uri} stdout: tool Bash: npm test`, - `agent planner ${BUNDLE.request.uri} stdout: tool Write: /tmp/plan.md`, - `agent planner ${BUNDLE.request.uri} stdout: result: Done`, - `turn ${BUNDLE.request.uri} exited without an artifact or question; marked crashed`, - ]) + assert.deepEqual(messages.slice(1), [`turn ${BUNDLE.request.uri} exited without an artifact or question; marked crashed`]) }) }) @@ -436,9 +427,8 @@ it('stamps the latest model the agent reports, including a mid-turn switch', asy log: (message) => messages.push(message), }) assert.match(messages[0], /model opus$/) - assert.deepEqual(messages.slice(1, 4), [ + assert.deepEqual(messages.slice(1, 3), [ `agent planner ${BUNDLE.request.uri} running on model claude-opus-4-8-20251101`, - `agent planner ${BUNDLE.request.uri} stdout: assistant: thinking`, `agent planner ${BUNDLE.request.uri} running on model claude-sonnet-4-8-20251101 (switched from claude-opus-4-8-20251101)`, ]) // One line per change only — the repeat on the second event does not log again. @@ -1307,7 +1297,7 @@ it('a managed-auth codex turn mounts a scratch home, redacts tokens, and writes passwd, `radial:x:${process.getuid?.() ?? 1000}:${process.getgid?.() ?? 1000}:Radial turn:/home/radial:/bin/sh\n`, ) - assert.ok(messages.some((message) => /agent .* stderr: \[redacted\]/.test(message))) + assert.ok(messages.every((message) => !message.includes(' stderr: '))) assert.ok(messages.every((message) => !message.includes('managed-refresh-secret'))) assert.equal( JSON.parse(await readFile(join(authDir, 'auth.json'), 'utf8')).tokens.refresh_token, @@ -1375,12 +1365,7 @@ it('any provider credential the daemon holds reaches the container, and secrets modelEnv: { OPENROUTER_API_KEY: 'sk-or-v1-secret-value', CLOUDFLARE_ACCOUNT_ID: 'acct-1234' }, log: (message) => messages.push(message), }) - // Redaction is by NAME: blanket-redacting an account id (or a region, or a cache setting) - // would mangle every log line that happened to contain it. - assert.equal( - messages[1], - `agent planner ${BUNDLE.request.uri} stdout: auth failed for [redacted] on acct-1234`, - ) + assert.ok(messages.every((message) => !message.includes('auth failed'))) }) }) @@ -1428,11 +1413,6 @@ it("a harness's rendering never reaches another harness's output", async () => { checkout: noopCheckout, log: (message) => messages.push(message), }) - assert.deepEqual(messages.slice(1, -1), [ - `agent planner ${BUNDLE.request.uri} running on model openai/gpt-5.2`, - `agent planner ${BUNDLE.request.uri} stdout: assistant: On it`, - `agent planner ${BUNDLE.request.uri} stdout: tool bash: pnpm test`, - `agent planner ${BUNDLE.request.uri} stdout: plain unstructured warning`, - ]) + assert.deepEqual(messages.slice(1, -1), [`agent planner ${BUNDLE.request.uri} running on model openai/gpt-5.2`]) }) })