From 881d264d5c2674e87d592068396ff860a5d65f4f Mon Sep 17 00:00:00 2001 From: cwiklik Date: Fri, 30 Oct 2026 11:36:10 -0400 Subject: [PATCH] fix(control-plane,mocactl): count completed turns, so sessions stop reporting 0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `turns` was on the runtime hash's allow-list (ownership.ts) and read by both the session summary (handlers.ts) and the resources view (resources.ts), but nothing ever wrote it: `runtimeFieldsForTurn` stamps turnStartedAt / turnEndedAt / lastTurnAt and no other putRuntime caller exists. So every real deployment reported 0 turns, visible in the mocactl TUI's Sessions overlay -- and #416 left the field out of `mocactl sessions` because of it. Counted with HINCRBY, not a read-modify-write. The server does not serialize concurrent turns of one session (ADR-0036), so reading the old value would have both turns read N and both write N+1, losing a count. This adds hIncrBy to CpRedisLike and a bumpTurns() that goes through the same guard as every other index write, so a transport failure is still redis_unavailable. `turns` is now readable but NOT settable through putRuntime (RUNTIME_COUNTERS). It is the only runtime field whose new value depends on its old one, and the tier that self-reports this hash is untrusted (spec §7.4) -- a `turns` passed to putRuntime would let a compromised harness claim any count. It is dropped on write, as a field off the allow-list is, rather than stored and then ignored. The reporter bumps when the fields carry turnEndedAt, rather than at each of the six call sites: a completed turn is exactly the report that stamps that timestamp, so no call site can report an end and forget to count it, and the counter cannot drift from the timestamp beside it. With a writer in place, `mocactl sessions` brings the field back: a TURNS column in the table and `turns` in `--json`, both straight from the server's summary. The local transcript store is deliberately not the source -- a headless `mocactl run` writes no transcript and a session resumed on another machine has none, yet both have run turns (#417's other half). The TUI already renders `s.turns` (Sessions.tsx), so it shows real counts with no change. Eight mutants, all caught: a read-modify-write instead of HINCRBY, a dropped RUNTIME_COUNTERS guard (turns forgeable through putRuntime), a reporter that never bumps, one that bumps unconditionally (counting started turns), a summary hardcoded to 0, a fake-redis op log that cannot tell HINCRBY from hSet, a TURNS column fed from the local transcript, and a header with no cell. Closes #420. Assisted-By: Claude (Anthropic AI) Signed-off-by: cwiklik --- packages/control-plane/src/ownership.ts | 26 ++++++++ .../test/handlers-sessions.test.ts | 16 +++++ .../control-plane/test/helpers/fake-redis.ts | 9 +++ packages/control-plane/test/ownership.test.ts | 45 ++++++++++++++ packages/knative-server/src/turn-auth.ts | 5 ++ .../knative-server/test/turn-auth.test.ts | 62 +++++++++++++++++++ packages/mocactl/src/headless.ts | 9 ++- packages/mocactl/test/headless-manage.test.ts | 33 ++++++++-- 8 files changed, 199 insertions(+), 6 deletions(-) diff --git a/packages/control-plane/src/ownership.ts b/packages/control-plane/src/ownership.ts index 5812d832..165168a1 100644 --- a/packages/control-plane/src/ownership.ts +++ b/packages/control-plane/src/ownership.ts @@ -13,6 +13,8 @@ import { sanitizeLimits, type ModelLimits } from './models.js'; */ export interface CpRedisLike { hSet(key: string, values: Record): Promise; + /** Atomic +N on one hash field, for the runtime hash's counters (see RUNTIME_COUNTERS). */ + hIncrBy(key: string, field: string, by: number): Promise; hGetAll(key: string): Promise>; zAdd(key: string, member: { score: number; value: string }): Promise; zRem(key: string, member: string): Promise; @@ -113,6 +115,18 @@ const RUNTIME_FIELDS = [ 'workspaceResetFrom', ] as const; +/** + * Readable, but NOT writable through `putRuntime` -- the one asymmetry in the runtime hash. + * + * `turns` is a counter, so it is the only runtime field whose new value depends on its old one. Two + * things follow. It cannot ride the merging `hSet`: the server does not serialize concurrent turns + * of one session (ADR-0036), so a read-modify-write would have both turns read N and both write + * N+1. And it must not be settable to an arbitrary value by the tier that self-reports this hash + * (spec §7.4), which a `fields.turns` passed to `putRuntime` would be. `bumpTurns` -- a server-side + * HINCRBY of exactly +1 -- is the only writer; everything else is dropped on write. + */ +const RUNTIME_COUNTERS: readonly string[] = ['turns']; + /** A stored limits field, read tolerantly: '' or a corrupted value is "no limits", never a throw. */ function readLimits(raw: string | undefined): ModelLimits | undefined { if (!raw) return undefined; @@ -264,12 +278,24 @@ export class OwnershipIndex { async putRuntime(sid: string, fields: Record): Promise { const allowed: Record = {}; for (const f of RUNTIME_FIELDS) { + // A counter is readable but never settable here -- see RUNTIME_COUNTERS. Dropped on write, + // like a field off the allow-list, rather than stored and then ignored. + if (RUNTIME_COUNTERS.includes(f)) continue; if (fields[f] !== undefined) allowed[f] = fields[f]!; } if (Object.keys(allowed).length === 0) return; await this.guard(() => this.redis.hSet(runtimeKey(sid), allowed)); } + /** + * Count one completed turn (#420). HINCRBY, not a read-modify-write through `putRuntime`: the + * server does not serialize concurrent turns of one session, so reading the old value here would + * undercount. See RUNTIME_COUNTERS for why this is the only writer of `turns`. + */ + async bumpTurns(sid: string): Promise { + await this.guard(() => this.redis.hIncrBy(runtimeKey(sid), 'turns', 1)); + } + async getRuntime(sid: string): Promise> { const h = await this.guard(() => this.redis.hGetAll(runtimeKey(sid))); const out: Record = {}; diff --git a/packages/control-plane/test/handlers-sessions.test.ts b/packages/control-plane/test/handlers-sessions.test.ts index bf776195..bed83074 100644 --- a/packages/control-plane/test/handlers-sessions.test.ts +++ b/packages/control-plane/test/handlers-sessions.test.ts @@ -455,6 +455,22 @@ describe('GET and DELETE /v1/sessions/{id}, POST .../token', () => { }); }); + it("grows the summary's turn count as turns complete, rather than reporting 0 forever (#420)", async () => { + // `turns` was read here and allow-listed for the runtime hash, but no writer existed, so every + // real deployment reported 0. This is the end-to-end shape the issue asks for: what the data + // plane's reporter calls on turn end must be what this summary shows. + const summary = async () => + ( + (await HANDLERS.getSession!(ctx({ principal: alice, params: { id: 'sid-fixed' } }), d)) + .body as { turns: number } + ).turns; + expect(await summary()).toBe(0); + await d.index.bumpTurns('sid-fixed'); + expect(await summary()).toBe(1); + await d.index.bumpTurns('sid-fixed'); + expect(await summary()).toBe(2); + }); + it('re-mints a session token, because a session outlives a 5-minute token', async () => { const res = await HANDLERS.mintSessionToken!( ctx({ principal: alice, params: { id: 'sid-fixed' } }), diff --git a/packages/control-plane/test/helpers/fake-redis.ts b/packages/control-plane/test/helpers/fake-redis.ts index f19c758f..37bf9110 100644 --- a/packages/control-plane/test/helpers/fake-redis.ts +++ b/packages/control-plane/test/helpers/fake-redis.ts @@ -17,6 +17,15 @@ export function fakeRedis() { hashes.set(key, { ...(hashes.get(key) ?? {}), ...values }); return 1; }, + async hIncrBy(key, field, by) { + // The field name is recorded too: `turns` must move by HINCRBY and never by the merging hSet, + // and an op log that said only `hIncrBy ` could not tell those apart. + ops.push(`hIncrBy ${key} ${field}`); + const h = hashes.get(key) ?? {}; + const next = (Number(h[field]) || 0) + by; + hashes.set(key, { ...h, [field]: String(next) }); + return next; + }, async hGetAll(key) { return { ...(hashes.get(key) ?? {}) }; }, diff --git a/packages/control-plane/test/ownership.test.ts b/packages/control-plane/test/ownership.test.ts index b7d7a77f..ff63e538 100644 --- a/packages/control-plane/test/ownership.test.ts +++ b/packages/control-plane/test/ownership.test.ts @@ -281,6 +281,51 @@ describe('runtime hash', () => { }); }); +describe('bumpTurns', () => { + it('counts a completed turn, so the session summary stops reporting 0 (#420)', async () => { + const f = fakeRedis(); + const index = new OwnershipIndex(f.redis); + await index.bumpTurns('sid-1'); + await index.bumpTurns('sid-1'); + expect((await index.getRuntime('sid-1')).turns).toBe('2'); + }); + + it('increments server-side, so concurrent turns of one session cannot lose a count', async () => { + // The server does not serialize concurrent turns of one session (ADR-0036), so a + // read-modify-write through putRuntime would have both turns read N and both write N+1. + // HINCRBY is the whole point: assert on the command, since a fake that merely ends at the + // right number would also pass for a racy implementation run sequentially. + const f = fakeRedis(); + const index = new OwnershipIndex(f.redis); + await Promise.all([index.bumpTurns('sid-1'), index.bumpTurns('sid-1')]); + expect((await index.getRuntime('sid-1')).turns).toBe('2'); + expect(f.ops).toEqual([ + `hIncrBy ${runtimeKey('sid-1')} turns`, + `hIncrBy ${runtimeKey('sid-1')} turns`, + ]); + // Never through the merging hSet path. + expect(f.ops.some((o) => o.startsWith('hSet'))).toBe(false); + }); + + it('is the only way turns is written, so a compromised harness cannot set it to any value', async () => { + // `turns` is on RUNTIME_FIELDS so getRuntime returns it, but the harness self-reports the + // runtime hash (spec §7.4). Were it writable through putRuntime, a compromised brain tier could + // claim any count. Only the +1 here may move it. + const f = fakeRedis(); + const index = new OwnershipIndex(f.redis); + await index.bumpTurns('sid-1'); + await index.putRuntime('sid-1', { harnessPod: 'h-1', turns: '9999' }); + expect(f.hashes.get(runtimeKey('sid-1'))).not.toHaveProperty('turns', '9999'); + expect((await index.getRuntime('sid-1')).turns).toBe('1'); + }); + + it('turns a Redis transport failure into redis_unavailable, like every other index write', async () => { + const down = () => Promise.reject(new Error('connect ECONNREFUSED')); + const index = new OwnershipIndex({ ...fakeRedis().redis, hIncrBy: down } as never); + await expect(index.bumpTurns('sid-1')).rejects.toMatchObject({ code: 'redis_unavailable' }); + }); +}); + describe('audit', () => { it('caps the audit stream with an approximate MAXLEN on every write', async () => { const f = fakeRedis(); diff --git a/packages/knative-server/src/turn-auth.ts b/packages/knative-server/src/turn-auth.ts index 7936143d..a971c8fd 100644 --- a/packages/knative-server/src/turn-auth.ts +++ b/packages/knative-server/src/turn-auth.ts @@ -463,6 +463,11 @@ export function makeRuntimeReporter( } await ready; await index?.putRuntime(sessionId, fields); + // A completed turn is exactly the report that stamps turnEndedAt, so the count rides that + // rather than a flag of its own: no call site can report an end and forget to count it, and + // the counter cannot drift from the timestamp beside it (#420). Separate from the putRuntime + // above because it is an HINCRBY -- `turns` is never settable through the merging hSet. + if (fields.turnEndedAt !== undefined) await index?.bumpTurns(sessionId); } catch { // A transient failure (e.g. a Redis restart mid-rollout) must not permanently disable this // reporter for the rest of the process's life: clear the memoised state so the NEXT call diff --git a/packages/knative-server/test/turn-auth.test.ts b/packages/knative-server/test/turn-auth.test.ts index 2fb589c8..f7a7135a 100644 --- a/packages/knative-server/test/turn-auth.test.ts +++ b/packages/knative-server/test/turn-auth.test.ts @@ -838,6 +838,68 @@ describe('makeRuntimeReporter', () => { expect(hSet).toHaveBeenCalled(); }); }); + + describe('counts completed turns (#420)', () => { + // `turns` was on the control plane's runtime allow-list and read by both the session summary + // and the resources view, but NOTHING ever wrote it, so every deployment reported 0 turns. + const client = () => { + const hSet = vi.fn(async () => undefined); + const hIncrBy = vi.fn(async () => 1); + vi.mocked(createClient).mockImplementation( + () => + ({ + on: vi.fn(), + connect: vi.fn(async () => undefined), + hSet, + hIncrBy, + destroy: vi.fn(), + }) as unknown as ReturnType, + ); + return { hSet, hIncrBy }; + }; + + it('increments once per turn end, with HINCRBY rather than a racy read-modify-write', async () => { + const { hIncrBy } = client(); + await makeRuntimeReporter('redis://127.0.0.1:6379')('sid-1', { + turnEndedAt: '1757000000000', + lastTurnAt: '1757000000000', + }); + expect(hIncrBy).toHaveBeenCalledTimes(1); + expect(hIncrBy).toHaveBeenCalledWith('sh:cp:session:sid-1:runtime', 'turns', 1); + }); + + it('does not count a turn that has only started, so an in-flight turn is not already counted', async () => { + const { hIncrBy, hSet } = client(); + await makeRuntimeReporter('redis://127.0.0.1:6379')('sid-1', { + turnStartedAt: '1757000000000', + }); + expect(hSet).toHaveBeenCalledTimes(1); + expect(hIncrBy).not.toHaveBeenCalled(); + }); + + it('still reports the end timestamps, so the counter does not displace the fields it rides with', async () => { + const { hSet, hIncrBy } = client(); + await makeRuntimeReporter('redis://127.0.0.1:6379')('sid-1', { + turnEndedAt: '1757000000000', + lastTurnAt: '1757000000000', + sandboxId: 'c-0', + }); + expect(hSet).toHaveBeenCalledTimes(1); + expect(hSet.mock.calls[0]).toMatchObject([ + 'sh:cp:session:sid-1:runtime', + { turnEndedAt: '1757000000000', lastTurnAt: '1757000000000', sandboxId: 'c-0' }, + ]); + expect(hIncrBy).toHaveBeenCalledTimes(1); + }); + + it('never fails the turn when the increment fails: display-only data', async () => { + const { hIncrBy } = client(); + hIncrBy.mockRejectedValue(new Error('connection lost')); + await expect( + makeRuntimeReporter('redis://127.0.0.1:6379')('sid-1', { turnEndedAt: '1757000000000' }), + ).resolves.toBeUndefined(); + }); + }); }); describe('configRef from the exchange', () => { diff --git a/packages/mocactl/src/headless.ts b/packages/mocactl/src/headless.ts index 3436407d..c9fae83b 100644 --- a/packages/mocactl/src/headless.ts +++ b/packages/mocactl/src/headless.ts @@ -600,13 +600,15 @@ export async function cmdSessions(rt: Runtime, io: Io, opts: ManageOptions): Pro return failed(io, err); } // Titles are local (renamed in the TUI, or the first prompt); the control plane keeps none. - // No turn count: the control plane reports one, but nothing writes it yet, so it is always 0. + // `turns` is the server's own count of completed turns, left out when #406 added this command + // because nothing wrote it then and it read 0 on every deployment (#420 gave it a writer). const rows = sessions.map((s) => ({ sessionId: s.sessionId, title: rt.transcripts?.load(s.sessionId)?.title ?? null, state: s.state, createdAt: s.createdAt, lastTurnAt: s.lastTurnAt, + turns: s.turns, })); if (opts.json) { io.out(JSON.stringify({ sessions: rows }) + '\n'); @@ -616,11 +618,14 @@ export async function cmdSessions(rt: Runtime, io: Io, opts: ManageOptions): Pro io.err('no sessions'); return 0; } - const head = ['ID', 'STATE', 'CREATED (UTC)', 'LAST TURN (UTC)', 'TITLE']; + const head = ['ID', 'STATE', 'TURNS', 'CREATED (UTC)', 'LAST TURN (UTC)', 'TITLE']; const body = rows.map((r) => [ r.sessionId, r.state, + // Stringified here rather than left to `table`, so it passes through the same sanitizeRemote + // as every other cell: this is a remote value, and `turns: number` is only what the type says. + String(r.turns), utc(r.createdAt), r.lastTurnAt === null ? '-' : utc(r.lastTurnAt), r.title ?? '', diff --git a/packages/mocactl/test/headless-manage.test.ts b/packages/mocactl/test/headless-manage.test.ts index df0dafed..df049f5a 100644 --- a/packages/mocactl/test/headless-manage.test.ts +++ b/packages/mocactl/test/headless-manage.test.ts @@ -44,7 +44,10 @@ describe('cmdSessions', () => { const listSessions = vi.fn(async (opts?: { cursor?: number }) => opts?.cursor === undefined ? { sessions: [summary('s-1')], nextCursor: 7 } - : { sessions: [summary('s-2', { lastTurnAt: null, state: 'deleting' })], nextCursor: null }, + : { + sessions: [summary('s-2', { lastTurnAt: null, state: 'deleting', turns: 0 })], + nextCursor: null, + }, ); const rt = testRuntime({ cp: fakeControlPlane({ listSessions }) }); rt.transcripts!.rename('s-1', 'fix \u001b]52;c;aGk=\u0007the build'); @@ -52,11 +55,12 @@ describe('cmdSessions', () => { expect(await cmdSessions(rt, o, { json: false })).toBe(0); expect(listSessions.mock.calls.map((c) => c[0]?.cursor)).toEqual([undefined, 7]); const lines = o.stdout.trimEnd().split('\n'); - expect(lines[0]).toMatch(/^ID\s+STATE\s+CREATED \(UTC\)\s+LAST TURN \(UTC\)\s+TITLE$/); + expect(lines[0]).toMatch(/^ID\s+STATE\s+TURNS\s+CREATED \(UTC\)\s+LAST TURN \(UTC\)\s+TITLE$/); expect(lines[1]).toMatch( - /^s-1\s+active\s+2026-09-30 09:15\s+2026-09-30 09:20\s+fix the build$/, + /^s-1\s+active\s+2\s+2026-09-30 09:15\s+2026-09-30 09:20\s+fix the build$/, ); - expect(lines[2]).toMatch(/^s-2\s+deleting\s+2026-09-30 09:15\s+-$/); + // A session that has run no turn reads 0, not a blank: the count is always known (#420). + expect(lines[2]).toMatch(/^s-2\s+deleting\s+0\s+2026-09-30 09:15\s+-$/); }); it('prints the documented --json shape', async () => { @@ -79,6 +83,7 @@ describe('cmdSessions', () => { state: 'active', createdAt: Date.UTC(2026, 8, 30, 9, 15), lastTurnAt: Date.UTC(2026, 8, 30, 9, 20), + turns: 2, }, { sessionId: 's-2', @@ -86,12 +91,32 @@ describe('cmdSessions', () => { state: 'active', createdAt: Date.UTC(2026, 8, 30, 9, 15), lastTurnAt: Date.UTC(2026, 8, 30, 9, 20), + turns: 2, }, ], }); expect(o.stdout.endsWith('\n')).toBe(true); }); + it("reports the SERVER's turn count, not the local transcript's (#420)", async () => { + // The field was left out of this command when #406 added it, because nothing wrote it and it + // read 0 everywhere; #420 gave it a writer. It must come from the summary: the local transcript + // store is the one thing that is NOT authoritative here -- a headless run writes no transcript, + // and a session resumed on another machine has none at all, yet both have run turns. + const rt = testRuntime({ + cp: fakeControlPlane({ + listSessions: async () => ({ sessions: [summary('s-1', { turns: 7 })], nextCursor: null }), + }), + }); + expect(rt.transcripts!.load('s-1')).toBeNull(); + const o = io(); + expect(await cmdSessions(rt, o, { json: true })).toBe(0); + expect((JSON.parse(o.stdout) as { sessions: { turns: number }[] }).sessions[0]!.turns).toBe(7); + const t = io(); + expect(await cmdSessions(rt, t, { json: false })).toBe(0); + expect(t.stdout.trimEnd().split('\n')[1]).toMatch(/^s-1\s+active\s+7\s/); + }); + it('keeps stdout empty with no sessions, and prints an empty list with --json', async () => { const o = io(); expect(await cmdSessions(testRuntime(), o, { json: false })).toBe(0);