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..4407db2 100644 --- a/worker/src/store.ts +++ b/worker/src/store.ts @@ -137,6 +137,13 @@ 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[]] { + // 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. */ export interface RoomScope { identity: string; links: string[]; groups: string[] } @@ -172,7 +179,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 +319,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 +443,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 +483,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..0998576 100644 --- a/worker/test/defrag.test.ts +++ b/worker/test/defrag.test.ts @@ -228,3 +228,22 @@ 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"); + // 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"); +});