worker: hide other sessions' unjudged candidates when S1 is wired

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

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

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<string> {
@@ -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<Claim[]> {
+    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<Claim[]> {
+    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);
   }
 

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");
+});