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);