diff --git a/src/api/app-sessions.ts b/src/api/app-sessions.ts index 6faaf7f23..4bf62a7f9 100644 --- a/src/api/app-sessions.ts +++ b/src/api/app-sessions.ts @@ -44,6 +44,7 @@ export function createSessionMethods( | "updateSession" | "regenerateTitle" | "spawnSession" + | "discardSession" | "forkSession" | "grant" | "revokeGrant" @@ -518,6 +519,22 @@ export function createSessionMethods( return create(); }, + async discardSession(sessionId, principalId) { + const session = await deps.sessions.get(sessionId); + if (!session) return false; + const mine = await deps.sessions.listByParticipant(principalId); + if (!mine.some((s) => s.id === sessionId)) return false; + if (!(await deps.sessions.deleteSessionIfEmpty(sessionId))) return false; + deps.auditLog.record({ + at: Date.now(), + principalId, + action: "session.discard", + resource: sessionId, + scopeLabel: session.scopeId, + }); + return true; + }, + async grant(g) { await deps.acl.grant(g, await artifactAuthor(g.ownerScopeId, g.ref)); deps.auditLog.record({ diff --git a/src/api/app-types.ts b/src/api/app-types.ts index 679567dc7..ae62386f3 100644 --- a/src/api/app-types.ts +++ b/src/api/app-types.ts @@ -281,6 +281,7 @@ export interface App { ): Promise; regenerateTitle(sessionId: string, principalId: string): Promise<{ title: string | null } | null>; spawnSession(principalId: string, opts: { scopeId: ScopeId; title?: string }): Promise<{ session: Session } | null>; + discardSession(sessionId: string, principalId: string): Promise; forkSession( sessionId: string, principalId: string, diff --git a/src/api/routes/surface.ts b/src/api/routes/surface.ts index 3fbe270af..aba0e13e7 100644 --- a/src/api/routes/surface.ts +++ b/src/api/routes/surface.ts @@ -103,18 +103,27 @@ async function spawnAgentConversation(ctx: ApiCtx): Promise { }); if (!out) return sendJson(res, 404, { error: "not_found", message: "cannot start a session in this scope" }); const session = out.session; + const sessionScope = parseScopeId(session.scopeId); const turn = await app.turn({ surface: session.surface ?? "web", actor: { externalId: capability.actorId }, conversation: { kind: session.type, threadRef: session.threadRef, + ...(sessionScope.kind === "channel" || sessionScope.kind === "group" ? { channelRef: sessionScope.ref } : {}), ...(session.channelName ? { channelName: session.channelName } : {}), }, text: b.text, spawned: true, async: true, }); + if (turn.status === "refused") { + await app.discardSession(session.id, capability.actorId); + return sendJson(res, 409, { + error: "seed_turn_refused", + message: (turn as { reason?: string }).reason ?? "the first message was refused", + }); + } const runId = (turn as { runId?: string }).runId; return sendJson(res, 202, { session, turn: { status: turn.status, ...(runId ? { runId } : {}) } }); } diff --git a/src/sessions/memory-session-store.ts b/src/sessions/memory-session-store.ts index f1f1986ce..8869b5c68 100644 --- a/src/sessions/memory-session-store.ts +++ b/src/sessions/memory-session-store.ts @@ -107,6 +107,7 @@ export function createMemorySessionStore(opts: StoreOptions = {}): SessionStore }, async acquireLease(sessionId, holder): Promise { + if (!sessions.has(sessionId)) return { lease: null }; const held = leases.get(sessionId); if (held && now() < held.expiresAt) return { @@ -143,6 +144,15 @@ export function createMemorySessionStore(opts: StoreOptions = {}): SessionStore leases.delete(sessionId); }, + async deleteSessionIfEmpty(sessionId) { + if (!sessions.has(sessionId)) return false; + if ((entries.get(sessionId)?.length ?? 0) > 0) return false; + const held = leases.get(sessionId); + if (held && now() < held.expiresAt) return false; + await this.deleteSession(sessionId); + return true; + }, + async forceReleaseLease(sessionId) { leases.delete(sessionId); }, diff --git a/src/sessions/postgres-session-store.ts b/src/sessions/postgres-session-store.ts index 75942126b..a6917cfbd 100644 --- a/src/sessions/postgres-session-store.ts +++ b/src/sessions/postgres-session-store.ts @@ -262,8 +262,12 @@ export function createPostgresSessionStore(connectionString: string, opts: Store OR ${lastActivityExpr("s")} > (EXTRACT(EPOCH FROM now()) * 1000)::bigint - 172800000`, ]); + const lockSession = (client: PoolClient, sessionId: string) => + client.query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))", [sessionId]); + const withLease = async (lease: Lease, invalidMsg: string, fn: (client: PoolClient) => Promise): Promise => withPgTransaction(await pool(), async (client) => { + await lockSession(client, lease.sessionId); const held = await client.query("SELECT token FROM session_leases WHERE session_id = $1 FOR UPDATE", [ lease.sessionId, ]); @@ -371,27 +375,31 @@ export function createPostgresSessionStore(connectionString: string, opts: Store async acquireLease(sessionId, holder): Promise { const token = randomUUID(); const t = now(); - const rows = await q( - `INSERT INTO session_leases(session_id, token, expires_at, holder, acquired_at) - VALUES ($1,$2,$3,$5,$4) - ON CONFLICT (session_id) DO UPDATE - SET token = $2, expires_at = $3, holder = $5, acquired_at = $4 - WHERE session_leases.expires_at <= $4 - RETURNING token`, - [sessionId, token, t + leaseTtlMs, t, holder ?? null], - ); - if (rows[0]) return { lease: { sessionId, token } }; - const held = await q("SELECT expires_at, holder, acquired_at FROM session_leases WHERE session_id = $1", [ - sessionId, - ]); - const row = held[0]; - if (!row) return { lease: null }; - return { - lease: null, - ...(row.holder != null ? { heldBy: row.holder as LeaseHolder } : {}), - ...(row.acquired_at != null ? { heldSince: Number(row.acquired_at) } : {}), - heldUntil: Number(row.expires_at), - }; + return withPgTransaction(await pool(), async (client) => { + await lockSession(client, sessionId); + const granted = await client.query( + `INSERT INTO session_leases(session_id, token, expires_at, holder, acquired_at) + SELECT $1, $2, $3, $5, $4 WHERE EXISTS (SELECT 1 FROM sessions WHERE id = $1) + ON CONFLICT (session_id) DO UPDATE + SET token = $2, expires_at = $3, holder = $5, acquired_at = $4 + WHERE session_leases.expires_at <= $4 + RETURNING token`, + [sessionId, token, t + leaseTtlMs, t, holder ?? null], + ); + if (granted.rows[0]) return { lease: { sessionId, token } }; + const held = await client.query( + "SELECT expires_at, holder, acquired_at FROM session_leases WHERE session_id = $1", + [sessionId], + ); + const row = held.rows[0]; + if (!row) return { lease: null }; + return { + lease: null, + ...(row.holder != null ? { heldBy: row.holder as LeaseHolder } : {}), + ...(row.acquired_at != null ? { heldSince: Number(row.acquired_at) } : {}), + heldUntil: Number(row.expires_at), + }; + }); }, async releaseLease(lease): Promise { @@ -600,6 +608,7 @@ export function createPostgresSessionStore(connectionString: string, opts: Store async deleteSession(sessionId): Promise { await withPgTransaction(await pool(), async (client) => { + await lockSession(client, sessionId); await client.query("DELETE FROM session_llm_requests WHERE session_id = $1", [sessionId]); await client.query("DELETE FROM session_leases WHERE session_id = $1", [sessionId]); await client.query("DELETE FROM participants WHERE session_id = $1", [sessionId]); @@ -609,6 +618,25 @@ export function createPostgresSessionStore(connectionString: string, opts: Store }); }, + async deleteSessionIfEmpty(sessionId): Promise { + return withPgTransaction(await pool(), async (client) => { + await lockSession(client, sessionId); + const gone = await client.query( + `DELETE FROM sessions + WHERE id = $1 + AND NOT EXISTS (SELECT 1 FROM session_entries WHERE session_id = $1) + AND NOT EXISTS (SELECT 1 FROM session_leases WHERE session_id = $1 AND expires_at > $2)`, + [sessionId, now()], + ); + if (gone.rowCount === 0) return false; + await client.query("DELETE FROM session_llm_requests WHERE session_id = $1", [sessionId]); + await client.query("DELETE FROM session_leases WHERE session_id = $1", [sessionId]); + await client.query("DELETE FROM participants WHERE session_id = $1", [sessionId]); + await client.query("DELETE FROM session_tape WHERE session_id = $1", [sessionId]); + return true; + }); + }, + async listByParticipant(principalId): Promise { const rows = await q( `SELECT s.*, p.title AS p_title, p.archived AS p_archived, p.pinned AS p_pinned, p.color AS p_color, diff --git a/src/sessions/session-store.ts b/src/sessions/session-store.ts index 6f056d525..89bd25497 100644 --- a/src/sessions/session-store.ts +++ b/src/sessions/session-store.ts @@ -440,6 +440,7 @@ export interface SessionStore { listByParticipant(principalId: string): Promise; deleteSession(sessionId: string): Promise; + deleteSessionIfEmpty(sessionId: string): Promise; updateParticipantView(sessionId: string, principalId: string, patch: ParticipantViewPatch): Promise; diff --git a/test/agent-conversations-route.test.ts b/test/agent-conversations-route.test.ts index ba06956d3..10eaadba9 100644 --- a/test/agent-conversations-route.test.ts +++ b/test/agent-conversations-route.test.ts @@ -85,6 +85,65 @@ describe("agent conversations self-API", async () => { ); }); + it("spawns a fresh channel conversation for a current member", async () => { + await built.app.upsertDirectory([{ principalId: "U1", displayName: "User One", type: "internal" }]); + await built.app.upsertChannels( + [{ channelId: "C1", name: "engineering", isPrivate: true }], + [{ channelId: "C1", principalId: "U1" }], + ); + const res = await post( + "/v1/conversations", + { text: "investigate the channel deployment" }, + await capFor("U1", scopeId("channel", "C1")), + ); + assert.equal(res.status, 202); + const body = (await res.json()) as { + session: { scopeId: string }; + turn: { status: string; runId?: string }; + }; + assert.equal(body.session.scopeId, scopeId("channel", "C1")); + assert.equal(body.turn.status, "queued"); + assert.ok(body.turn.runId); + const run = await built.runs.get(body.turn.runId); + assert.equal(run?.request.conversation.channelRef, "C1"); + }); + + it("discards the spawned session when the seed turn is refused (roster race)", async () => { + const racedApp: typeof built.app = { + ...built.app, + turn: async (req) => + (req as { spawned?: boolean }).spawned + ? { status: "refused", reason: "project membership changed; retry from the current project" } + : built.app.turn(req), + }; + const racedServer = createServer(racedApp, { signingSecret: SECRET }); + await new Promise((resolve) => racedServer.listen(0, resolve)); + const racedBase = `http://localhost:${(racedServer.address() as AddressInfo).port}`; + try { + const token = await capFor("U1"); + const listIds = async () => { + const listed = await get("/v1/conversations", token); + const { conversations } = (await listed.json()) as { conversations: Array<{ id: string }> }; + return conversations.map((c) => c.id).sort(); + }; + const before = await listIds(); + + const res = await fetch(`${racedBase}/v1/conversations`, { + method: "POST", + headers: { "content-type": "application/json", "x-agent-capability": token }, + body: JSON.stringify({ text: "seed that will be refused" }), + }); + assert.equal(res.status, 409); + const body = (await res.json()) as { error: string; message: string }; + assert.equal(body.error, "seed_turn_refused"); + assert.match(body.message, /membership changed/); + + assert.deepEqual(await listIds(), before, "no orphaned empty session survives the refused seed"); + } finally { + await new Promise((resolve) => racedServer.close(() => resolve())); + } + }); + it("spawn requires text and a capability", async () => { assert.equal((await post("/v1/conversations", { text: "hi" })).status, 401); assert.equal((await post("/v1/conversations", {}, await capFor("U1"))).status, 400); diff --git a/test/postgres-store.test.ts b/test/postgres-store.test.ts index 2a3657b11..cc5103bb1 100644 --- a/test/postgres-store.test.ts +++ b/test/postgres-store.test.ts @@ -287,7 +287,10 @@ test("pg deleteSession: hard-removes the session and its rows, leaves others", { const keptRow = (await s.listByParticipant("DEL1")).find((x) => x.id === keep.id); assert.ok(keptRow, "other sessions untouched"); assert.equal(keptRow!.hasEntries, false, "an entry-less session reads back hasEntries=false"); - assert.notEqual((await s.acquireLease(a.id)).lease, null, "lease row cleared (re-acquirable)"); + assert.equal((await s.acquireLease(a.id)).lease, null, "no lease is granted on a deleted session"); + const reborn = await s.getOrCreateByThread("del-t1", "dm", scope); + assert.notEqual(reborn.id, a.id, "the thread gets a fresh session"); + assert.notEqual((await s.acquireLease(reborn.id)).lease, null, "the fresh session is leasable"); }); test("pg scopeSessionSummaries: counts via aggregate, not per-transcript reads", { skip }, async () => { @@ -1180,3 +1183,104 @@ test("pg participant view: pin/color are per-participant, survive re-add, and cl assert.ok(!cleared.pinned); assert.equal(cleared.color ?? null, null, "null clears the color"); }); + +test( + "pg deleteSessionIfEmpty: an expired lease forfeits, and the stale holder cannot orphan entries", + { skip }, + async () => { + const s = createPostgresSessionStore(URL!, { leaseTtlMs: 40 }); + const scope = scopeId("personal", "USTALE"); + const sess = await s.getOrCreateByThread("web:USTALE:seed", "dm", scope); + const att = await s.acquireLease(sess.id); + assert.ok(att.lease); + await new Promise((r) => setTimeout(r, 60)); + assert.equal(await s.deleteSessionIfEmpty(sess.id), true, "an expired lease does not block the discard"); + assert.equal(await s.get(sess.id), null); + await assert.rejects( + s.append(att.lease!, { type: "user", payload: {}, scopeLabel: scope }), + /valid session lease/, + "the stale holder cannot append after the discard", + ); + const pg = (await import("pg")).default; + const raw = new pg.Pool({ connectionString: URL }); + const leftovers = await raw.query( + "SELECT (SELECT COUNT(*) FROM session_entries WHERE session_id = $1) AS e, (SELECT COUNT(*) FROM session_leases WHERE session_id = $1) AS l", + [sess.id], + ); + await raw.end(); + assert.equal(Number(leftovers.rows[0].e), 0, "no orphaned entries after the stale append is refused"); + assert.equal(Number(leftovers.rows[0].l), 0, "no orphaned lease row survives"); + }, +); + +test("pg deleteSession: racing a fresh lease acquisition never orphans entries", { skip }, async () => { + const s = createPostgresSessionStore(URL!); + const scope = scopeId("personal", "UHARD"); + for (let i = 0; i < 15; i++) { + const sess = await s.getOrCreateByThread(`web:UHARD:${i}`, "dm", scope); + const [attempt] = await Promise.all([s.acquireLease(sess.id), s.deleteSession(sess.id)]); + assert.equal(await s.get(sess.id), null, "the session is gone either way"); + if (attempt.lease) { + await assert.rejects( + s.append(attempt.lease, { type: "user", payload: {}, scopeLabel: scope }), + /valid session lease/, + "a lease granted before the delete cannot orphan entries", + ); + } + } + const pg = (await import("pg")).default; + const raw = new pg.Pool({ connectionString: URL }); + const leftovers = await raw.query( + "SELECT COUNT(*) AS n FROM session_leases l WHERE NOT EXISTS (SELECT 1 FROM sessions ss WHERE ss.id = l.session_id)", + ); + await raw.end(); + assert.equal(Number(leftovers.rows[0].n), 0, "no lease row survives its session"); +}); + +test("pg deleteSessionIfEmpty: racing a fresh lease acquisition never orphans the lease", { skip }, async () => { + const s = createPostgresSessionStore(URL!); + const scope = scopeId("personal", "URACE"); + for (let i = 0; i < 15; i++) { + const sess = await s.getOrCreateByThread(`web:URACE:${i}`, "dm", scope); + const [attempt, discarded] = await Promise.all([s.acquireLease(sess.id), s.deleteSessionIfEmpty(sess.id)]); + if (discarded) { + assert.equal(attempt.lease, null, "a discarded session never grants a lease"); + assert.equal(await s.get(sess.id), null); + } else { + assert.ok(attempt.lease, "when the discard is refused the lease was granted"); + assert.ok(await s.get(sess.id), "the session survives when the lease won"); + await s.releaseLease(attempt.lease!); + assert.equal(await s.deleteSessionIfEmpty(sess.id), true); + } + } +}); + +test("pg deleteSessionIfEmpty: a held lease or landed entries refuse the discard atomically", { skip }, async () => { + const s = createPostgresSessionStore(URL!); + const scope = scopeId("personal", "UDISC"); + const sess = await s.getOrCreateByThread("web:UDISC:seed", "dm", scope); + const att = await s.acquireLease(sess.id); + assert.ok(att.lease); + assert.equal(await s.deleteSessionIfEmpty(sess.id), false, "a held lease blocks the discard"); + await s.append(att.lease, { type: "user", payload: { text: "seed" }, scopeLabel: scope }); + await s.releaseLease(att.lease); + assert.equal(await s.deleteSessionIfEmpty(sess.id), false, "entries block the discard"); + assert.ok(await s.get(sess.id), "the refused discard leaves the session intact"); + + const empty = await s.getOrCreateByThread("web:UDISC:seed2", "dm", scope); + const att2 = await s.acquireLease(empty.id); + assert.ok(att2.lease); + await s.releaseLease(att2.lease); + assert.equal(await s.deleteSessionIfEmpty(empty.id), true, "a released empty session is discarded"); + assert.equal(await s.get(empty.id), null); + + const pg = (await import("pg")).default; + const raw = new pg.Pool({ connectionString: URL }); + const orphans = await raw.query( + "SELECT (SELECT COUNT(*) FROM session_entries WHERE session_id = $1) AS e, (SELECT COUNT(*) FROM session_leases WHERE session_id = $1) AS l", + [empty.id], + ); + await raw.end(); + assert.equal(Number(orphans.rows[0].e), 0, "no orphaned entries survive the discard"); + assert.equal(Number(orphans.rows[0].l), 0, "no orphaned lease survives the discard"); +}); diff --git a/test/session-store.test.ts b/test/session-store.test.ts index 42855ce56..38797944d 100644 --- a/test/session-store.test.ts +++ b/test/session-store.test.ts @@ -587,7 +587,10 @@ for (const [name, make] of backends) { true, "other sessions untouched", ); - assert.notEqual((await store.acquireLease(s.id)).lease, null, "lease row cleared (re-acquirable)"); + assert.equal((await store.acquireLease(s.id)).lease, null, "no lease is granted on a deleted session"); + const reborn = await store.getOrCreateByThread("t1", "dm", scope); + assert.notEqual(reborn.id, s.id, "the thread gets a fresh session"); + assert.notEqual((await store.acquireLease(reborn.id)).lease, null, "the fresh session is leasable"); }); test(`${name}: listByParticipant sets lastActivityAt to the most recent user message`, async () => { @@ -834,3 +837,34 @@ test("cronIdOf and sessionOrigin agree on which threadRefs are crons", () => { assert.equal(cronIdOf("cron:abc:slot"), "abc"); assert.equal(cronIdOf("dm:D1"), null); }); + +test("deleteSessionIfEmpty refuses while a lease is held and after entries land", async () => { + const nowRef = { v: 10_000_000_000 }; + const store = createMemorySessionStore({ now: () => nowRef.v, leaseTtlMs: 50 }); + const scope = scopeId("personal", "U1"); + const s = await store.getOrCreateByThread("web:U1:seed", "dm", scope); + const { lease } = await store.acquireLease(s.id); + assert.ok(lease); + assert.equal(await store.deleteSessionIfEmpty(s.id), false, "a held lease blocks the discard"); + await store.append(lease, { type: "user", payload: { text: "seed" }, scopeLabel: scope }); + await store.releaseLease(lease); + assert.equal(await store.deleteSessionIfEmpty(s.id), false, "entries block the discard"); + const empty = await store.getOrCreateByThread("web:U1:seed2", "dm", scope); + const second = await store.acquireLease(empty.id); + assert.ok(second.lease); + await store.releaseLease(second.lease); + assert.equal(await store.deleteSessionIfEmpty(empty.id), true, "a released empty session is discarded"); + assert.equal(await store.get(empty.id), null); + assert.equal((await store.acquireLease(empty.id)).lease, null, "no lease is granted on a discarded session"); + + const abandoned = await store.getOrCreateByThread("web:U1:seed3", "dm", scope); + const stale = await store.acquireLease(abandoned.id); + assert.ok(stale.lease); + nowRef.v += 60; + assert.equal(await store.deleteSessionIfEmpty(abandoned.id), true, "an expired lease does not block the discard"); + await assert.rejects( + store.append(stale.lease, { type: "user", payload: {}, scopeLabel: scope }), + /valid session lease/, + "the stale holder cannot append after the discard", + ); +});