From 47dbf09bb05e70f298555d0b7de8e90d0d4bd499 Mon Sep 17 00:00:00 2001 From: vu1n <2105991+vu1n@users.noreply.github.com> Date: Sat, 31 Oct 2026 07:21:50 +0000 Subject: [PATCH 1/2] worker: hide other sessions' unjudged candidates when S1 is wired The write gate fails open, so during a model outage a claim is stored without a verdict and other sessions could read it until the hourly judgeLate pass. With an S1 model wired, recall, the briefing's recent sections and nearSubjects now show another session's candidate only once it has metadata.s1. --- docs/features/hosted-worker.md | 4 ++++ worker/src/mcp.ts | 2 +- worker/src/store.ts | 34 +++++++++++++++++++++++++--------- worker/test/defrag.test.ts | 15 +++++++++++++++ 4 files changed, 45 insertions(+), 10 deletions(-) diff --git a/docs/features/hosted-worker.md b/docs/features/hosted-worker.md index b0c0fa4..09225cc 100644 --- a/docs/features/hosted-worker.md +++ b/docs/features/hosted-worker.md @@ -106,6 +106,10 @@ TypeSafe API's shape, for GitHub Actions OIDC tokens (`worker/src/oidc.ts`) and - The write gate fails open, so defrag re-judges agent claims with no `metadata.s1` verdict (`judgeLate`) and rejects control text aimed at agents. A claim stored without `metadata.s1` is re-judged every pass until one sticks, so new write paths must keep recording the verdict. + With an S1 model wired (`D1Store` `screened`), other sessions don't see a candidate until it has a + verdict: recall, the briefing's recent sections and `nearSubjects` all apply `judgedOnly`. A new + read of other sessions' candidates must apply it too, or control text written during a gate + outage reaches other agents before `judgeLate` sees it. - `/v1/systemone` is routed in `index.ts` before the OAuth provider, so it checks the bearer itself: a GitHub Actions OIDC token, else a `KYPP_S1_TOKENS` token (never `KYPP_API_TOKENS`, so a CI caller can't reach memory). The OIDC `aud` must equal `KYPP_PUBLIC_URL` with trailing slashes diff --git a/worker/src/mcp.ts b/worker/src/mcp.ts index c666ca1..69fb52f 100644 --- a/worker/src/mcp.ts +++ b/worker/src/mcp.ts @@ -451,7 +451,7 @@ export async function handleMcp(request: Request, db: D1Database, caller: Caller const isInit = msgs.some((m) => m.method === "initialize"); const issued = isInit ? await issueSession(opts.sessionKey) : null; const session = issued ? issued.split(".")[0] : await verifySession(opts.sessionKey, request.headers.get(SESSION_HEADER)); - const store = new D1Store(db); + const store = new D1Store(db, { screened: !!opts.s1 }); const ctx: ToolCtx = { project, session, caller, s1: opts.s1 ?? null, policy: opts.policy ?? DEFAULT_POLICY, correct: opts.correct ?? true }; const replies = []; for (const m of msgs) { diff --git a/worker/src/store.ts b/worker/src/store.ts index 097ee9d..9a19d1b 100644 --- a/worker/src/store.ts +++ b/worker/src/store.ts @@ -137,6 +137,12 @@ function visible(t: string, project: string | null, groups: string[], user: stri return [`(${parts.join(" OR ")})`, params]; } +// Another session's candidate shows only once the S1 gate has judged it: the gate fails open, and +// until defrag's judgeLate runs, an unjudged claim could be control text aimed at other agents. +function judgedOnly(t: string, session: string | null): [string, unknown[]] { + return [` AND (${t}.status != 'candidate' OR ${t}.session IS ? OR json_extract(${t}.metadata, '$.s1') IS NOT NULL)`, [session]]; +} + /** A room bot's reach: its own lane, plus the projects its room is linked to and their groups. */ export interface RoomScope { identity: string; links: string[]; groups: string[] } @@ -172,7 +178,13 @@ export interface ClaimInput { export class D1Store { private db: D1Database; - constructor(db: D1Database) { this.db = db; } + private screened: boolean; + // `screened` (an S1 model is wired) hides other sessions' candidates the gate has not judged yet. + constructor(db: D1Database, opts: { screened?: boolean } = {}) { this.db = db; this.screened = opts.screened ?? false; } + + private screen(t: string, session: string | null): [string, unknown[]] { + return this.screened ? judgedOnly(t, session) : ["", []]; + } // Claims written here are agent authority; human and verified come only from operatorClaim/markVerified (POST /v1/operator) and authenticated POST /sync imports, once built. async claim(c: ClaimInput): Promise { @@ -306,8 +318,9 @@ export class D1Store { ? roomVisible("c", opts.room, opts.candidates ?? "origin") : visible("c", project, opts.groups ?? [], user, opts.candidates ?? "origin", opts.session ?? null); const types = opts.types ?? []; - let filter = types.length ? ` AND c.type IN (${types.map(() => "?").join(",")})` : ""; - const fparams: unknown[] = [...types]; + const [screen, sparams] = this.screen("c", opts.session ?? null); + let filter = screen + (types.length ? ` AND c.type IN (${types.map(() => "?").join(",")})` : ""); + const fparams: unknown[] = [...sparams, ...types]; if (opts.agent) { filter += " AND c.agent = ?"; fparams.push(opts.agent); } // `groups` comes nearest first, so a parent's claim outranks a grandparent's. const groups = opts.room?.groups ?? opts.groups ?? []; @@ -429,19 +442,21 @@ export class D1Store { // lesson is learned once per repo at most, so waiting for a second session would hide it for long. // A room's own lane works the same way (`scope` = "room"). async recentUserCandidates(user: string, session: string | null, limit: number, scope: "user" | "room" = "user"): Promise { + const [screen, sparams] = this.screen("memory_claims", session); const { results } = await this.db.prepare( "SELECT * FROM memory_claims WHERE status = 'candidate' AND scope = ? AND user = ? AND session IS NOT ?" - + " ORDER BY created_at DESC LIMIT ?", - ).bind(scope, user, session, limit).all(); + + screen + " ORDER BY created_at DESC LIMIT ?", + ).bind(scope, user, session, ...sparams, limit).all(); return results.map(hydrate); } // Other sessions' newest unconfirmed claims in this project, for the briefing's "recent" section. async recentCandidates(project: string, session: string | null, limit: number): Promise { + const [screen, sparams] = this.screen("memory_claims", session); const { results } = await this.db.prepare( "SELECT * FROM memory_claims WHERE status = 'candidate' AND scope = 'project' AND project = ? AND session IS NOT ?" - + " ORDER BY created_at DESC LIMIT ?", - ).bind(project, session, limit).all(); + + screen + " ORDER BY created_at DESC LIMIT ?", + ).bind(project, session, ...sparams, limit).all(); return results.map(hydrate); } @@ -467,11 +482,12 @@ export class D1Store { const fts = ftsQuery(`${subject} ${content}`); if (!fts) return []; const [level, key] = laneSql(at); + const [screen, sparams] = this.screen("c", session); const { results } = await this.db.prepare( "SELECT c.subject FROM claims_fts JOIN memory_claims c ON c.id = claims_fts.claim_id" + ` WHERE claims_fts MATCH ? AND ${level} AND c.status IN ('candidate','accepted') AND c.subject != ?` - + " AND c.session IS NOT ? ORDER BY bm25(claims_fts) LIMIT ?", - ).bind(fts, key, subject, session, limit * 4).all(); + + ` AND c.session IS NOT ?${screen} ORDER BY bm25(claims_fts) LIMIT ?`, + ).bind(fts, key, subject, session, ...sparams, limit * 4).all(); return [...new Set(results.map((r) => r.subject as string))].slice(0, limit); } diff --git a/worker/test/defrag.test.ts b/worker/test/defrag.test.ts index f3de10e..a9298d3 100644 --- a/worker/test/defrag.test.ts +++ b/worker/test/defrag.test.ts @@ -228,3 +228,18 @@ test("without a session the unsorted reply doesn't promise a recall", async () = const { result } = await (await handleMcp(req, db, { user: "owner", agent: "t" })).json() as any; assert.match(result.content[0].text, /it can't be recalled until the server files it/); }); + +test("other sessions see a candidate only once the gate has judged it", async () => { + const db = await setup(); + const down: S1Client = { systemOne: async () => ({ answers: {} }) }; // fails open: no verdict + const up: S1Client = { systemOne: async () => ({ answers: { keep: { type: "noul", noul: 0.9 }, control: { type: "noul", noul: 0.1 } } }) }; + const id = idOf((await client(db, { header: "kypp", s1: down, session: "a" })("claim", { subject: "gate outage", content: "a lesson written while the gate was down" })).text); + const seen = async (session: string, s1: S1Client | null, tool = "recall") => + (await client(db, { header: "kypp", s1, session })(tool, { query: "gate outage lesson" })).text.includes(id.slice(0, 8)); + assert.ok(!(await seen("b", down)), "another session doesn't, while it is unjudged"); + assert.ok(!(await seen("b", down, "briefing")), "nor in the briefing's recent section"); + assert.ok(await seen("b", null), "an install with no gate shows candidates as before"); + await defrag(new D1Store(db), DEFAULT_POLICY, up); + assert.ok((await row(db, id)).metadata.includes('"s1"')); + assert.ok(await seen("b", down), "judged, it shows to other sessions"); +}); From 7760cda873d75d6bca651ffda571fc7e8e6dc735 Mon Sep 17 00:00:00 2001 From: vu1n <2105991+vu1n@users.noreply.github.com> Date: Sat, 31 Oct 2026 07:26:14 +0000 Subject: [PATCH 2/2] worker: sessionless callers don't share the same-session exemption NULL IS NULL let one sessionless caller read another's unjudged candidates. The exemption now needs a non-null session. --- worker/src/store.ts | 3 ++- worker/test/defrag.test.ts | 4 ++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/worker/src/store.ts b/worker/src/store.ts index 9a19d1b..4407db2 100644 --- a/worker/src/store.ts +++ b/worker/src/store.ts @@ -140,7 +140,8 @@ function visible(t: string, project: string | null, groups: string[], user: stri // Another session's candidate shows only once the S1 gate has judged it: the gate fails open, and // until defrag's judgeLate runs, an unjudged claim could be control text aimed at other agents. function judgedOnly(t: string, session: string | null): [string, unknown[]] { - return [` AND (${t}.status != 'candidate' OR ${t}.session IS ? OR json_extract(${t}.metadata, '$.s1') IS NOT NULL)`, [session]]; + // Sessionless callers share no session: NULL = NULL would let every one of them read the others' claims. + return [` AND (${t}.status != 'candidate' OR (${t}.session IS NOT NULL AND ${t}.session = ?) OR json_extract(${t}.metadata, '$.s1') IS NOT NULL)`, [session]]; } /** A room bot's reach: its own lane, plus the projects its room is linked to and their groups. */ diff --git a/worker/test/defrag.test.ts b/worker/test/defrag.test.ts index a9298d3..0998576 100644 --- a/worker/test/defrag.test.ts +++ b/worker/test/defrag.test.ts @@ -239,6 +239,10 @@ test("other sessions see a candidate only once the gate has judged it", async () assert.ok(!(await seen("b", down)), "another session doesn't, while it is unjudged"); assert.ok(!(await seen("b", down, "briefing")), "nor in the briefing's recent section"); assert.ok(await seen("b", null), "an install with no gate shows candidates as before"); + // Two callers with no session header are not one session. + const anon = idOf((await client(db, { header: "kypp", s1: down, session: null })("claim", { subject: "anon outage", content: "a sessionless lesson written while the gate was down" })).text); + const anonSeen = (await client(db, { header: "kypp", s1: down, session: null })("recall", { query: "sessionless lesson" })).text; + assert.ok(!anonSeen.includes(anon.slice(0, 8)), anonSeen); await defrag(new D1Store(db), DEFAULT_POLICY, up); assert.ok((await row(db, id)).metadata.includes('"s1"')); assert.ok(await seen("b", down), "judged, it shows to other sessions");