Forward the session scope ref when spawning a shared-scope conversation

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

src/api/app-types.ts

@@ -281,6 +281,7 @@ export interface App {
   ): Promise<Session | null>;
   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<boolean>;
   forkSession(
     sessionId: string,
     principalId: string,

src/api/routes/surface.ts

@@ -103,18 +103,27 @@ async function spawnAgentConversation(ctx: ApiCtx): Promise<void> {
   });
   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 } : {}) } });
 }

src/sessions/memory-session-store.ts

@@ -107,6 +107,7 @@ export function createMemorySessionStore(opts: StoreOptions = {}): SessionStore
     },
 
     async acquireLease(sessionId, holder): Promise<LeaseAttempt> {
+      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);
     },

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 <T>(lease: Lease, invalidMsg: string, fn: (client: PoolClient) => Promise<T>): Promise<T> =>
     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<LeaseAttempt> {
       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<void> {
@@ -600,6 +608,7 @@ export function createPostgresSessionStore(connectionString: string, opts: Store
 
     async deleteSession(sessionId): Promise<void> {
       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<boolean> {
+      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<Session[]> {
       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,

src/sessions/session-store.ts

@@ -440,6 +440,7 @@ export interface SessionStore {
   listByParticipant(principalId: string): Promise<Session[]>;
 
   deleteSession(sessionId: string): Promise<void>;
+  deleteSessionIfEmpty(sessionId: string): Promise<boolean>;
 
   updateParticipantView(sessionId: string, principalId: string, patch: ParticipantViewPatch): Promise<void>;
 

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

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