feat(capy): add explicit OAuth reply permits for native bindings

docs/capy.md

@@ -87,8 +87,30 @@ Before mounting new work, add routes, callback verification, subscriptions and
 durable delivery.
 No production caller writes binding rows yet.
 
-OAuth `assertActor`/`pin` is unchanged. A future OAuth-bound Capy journey is a
-separate admission/parity change, not an alias-table bypass in this registry.
+## Explicit OAuth reply permits (internal, not mounted)
+
+Migration `0046_capy_grant_permits.sql` keeps reply authority separate from the
+native binding. Optional `bind_capy_thread(thread_id)` requires an existing bound
+thread owned by this grant's person, current owner/admin membership and ownership
+of the connected key. It creates one live permit for the exact grant and binding,
+snapshotting generation, immutable remote ID and name. Repeated binds return the
+existing binding with `already_bound: true`; they never create or transfer a
+binding or change the approved OAuth persona.
+
+Only `send_relay_message as=<binding agent>` may resolve this permit. Generic
+`assertActor`/`pin`, retirement, inbox, conversation, channel, launch and Desktop
+identity rules remain unchanged. Every send checks the live bearer and permit,
+canonical key/role/route and remote custody before and after verification and
+again after parent/address reads. Revoking the grant revokes its permits in the
+same transaction. Binding retirement or generation/identity changes fail closed.
+Revocation cannot retract a write already accepted by Relaycast.
+
+`unbind_capy_thread(thread_id)` revokes only this grant's reply permit; other
+grants and native inbound delivery retain their independent binding authority.
+The optional MCP module is not supplied by the application server and runtime
+capability remains false. Before mounting, measure the extra remote checks and
+complete Control-owned callback/subscription/durable delivery. Those messaging
+callbacks do not belong to the separate provider-file ingestion lane.
 
 ## Required next slices
 

migrations/0046_capy_grant_permits.sql

@@ -0,0 +1,23 @@
+-- A reply permit belongs to one OAuth grant, never to its base persona.
+-- Native binding custody and inbound delivery are independent of this permit.
+ALTER TABLE capy_bindings ADD CONSTRAINT capy_bindings_permit_owner
+  UNIQUE (id,organization_id,workspace_id,user_id);
+CREATE TABLE capy_grant_permits (
+  id uuid PRIMARY KEY,
+  organization_id uuid NOT NULL,
+  workspace_id uuid NOT NULL,
+  user_id uuid NOT NULL,
+  grant_id uuid NOT NULL,
+  binding_id uuid NOT NULL,
+  binding_generation integer NOT NULL CHECK (binding_generation > 0),
+  agent_remote_id text NOT NULL CHECK (length(agent_remote_id) > 0),
+  agent_name text NOT NULL CHECK (length(agent_name) > 0),
+  created_at timestamptz NOT NULL DEFAULT now(),
+  revoked_at timestamptz,
+  revoke_reason text,
+  FOREIGN KEY (grant_id,organization_id,workspace_id,user_id)
+    REFERENCES mcp_oauth_grants(id,organization_id,workspace_id,user_id) ON DELETE CASCADE,
+  FOREIGN KEY (binding_id,organization_id,workspace_id,user_id)
+    REFERENCES capy_bindings(id,organization_id,workspace_id,user_id) ON DELETE CASCADE
+);
+CREATE UNIQUE INDEX capy_grant_permits_live ON capy_grant_permits(grant_id,binding_id) WHERE revoked_at IS NULL;

src/capy/bindings.ts

@@ -3,12 +3,13 @@ import { decryptSecret, encryptSecret, hashSecret, type EncryptionKeyring } from
 import type { Database, SqlExecutor } from "../database.js";
 import { RelaycastCallError, type RelaycastClient, type RemoteAgent } from "../messaging/relaycast.js";
 import { HOSTED_SOURCE, ownerHash } from "../mcp/relay-participants.js";
+import type { McpOAuthConnection } from "../mcp/oauth/admission.js";
 import { lockOrganization, lockUser } from "../tenancy.js";
 import { CapyRuntimeError, type CapyRuntime } from "./runtime.js";
 
 /** Internal service, called only after person admission. Never pass model/user
- * supplied acting IDs here. Native integration must use !managed && !oauth,
- * exactly like Factory; this service does not extend OAuth assertActor/pin.
+ * supplied acting IDs here. Native integration uses !managed && !oauth.
+ * OAuth replies require a separate exact-grant permit; assertActor/pin stay unchanged.
  */
 export type CapyActor = { userId: string; organizationId: string; workspaceId: string };
 export class CapyBindingError extends Error {
@@ -29,6 +30,7 @@ export const capyAgentName = (threadId: string) => `capy-${hashSecret(threadId).
 const THREAD = /^jam_[A-Za-z0-9]{6,64}$/;
 const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
 const changed = () => new CapyBindingError(409, "capy_binding_changed");
+type Permit = { id: string; binding_id: string; binding_generation: number; agent_remote_id: string; agent_name: string };
 
 export class CapyBindings {
   private readonly now: () => Date;
@@ -238,6 +240,90 @@ export class CapyBindings {
     }
   }
 
+  /** Explicit self-bind adds only this grant's send permit to an existing
+   * native binding. It never creates/reassigns a binding or pins a persona.
+   */
+  async bindGrant(connection: McpOAuthConnection, threadId: string): Promise<CapyBindingView & { already_bound: true }> {
+    if (!THREAD.test(threadId)) throw new CapyBindingError(400,"capy_invalid_thread");
+    const actor = connection.access.principal;
+    return this.locked(actor, async tx => {
+      await connection.lock(tx);
+      const row = (await tx.query<Row>(`SELECT * FROM capy_bindings WHERE thread_id=$1
+        AND workspace_id=$2 AND organization_id=$3 AND user_id=$4 AND state='bound' FOR UPDATE`,
+      [threadId,actor.workspaceId,actor.organizationId,actor.userId])).rows[0];
+      if (!row) throw new CapyBindingError(404,"capy_binding_not_found");
+      const live = await this.current(tx,actor,row);
+      if (!live.token_ciphertext || !live.agent_remote_id || capyAgentName(threadId) !== live.agent_name) throw changed();
+      // A stale generation/name is not repaired by use. Only an explicit bind
+      // under this grant may replace its old permit with a fresh snapshot.
+      await tx.query(`UPDATE capy_grant_permits SET revoked_at=$3,revoke_reason='binding_changed'
+        WHERE grant_id=$1 AND binding_id=$2 AND revoked_at IS NULL
+          AND (binding_generation<>$4 OR agent_remote_id<>$5 OR agent_name<>$6)`,
+      [connection.access.grantId,live.id,this.now(),live.generation,live.agent_remote_id,live.agent_name]);
+      await tx.query(`INSERT INTO capy_grant_permits
+        (id,organization_id,workspace_id,user_id,grant_id,binding_id,binding_generation,agent_remote_id,agent_name,created_at)
+        VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
+        ON CONFLICT (grant_id,binding_id) WHERE revoked_at IS NULL DO NOTHING`,
+      [randomUUID(),actor.organizationId,actor.workspaceId,actor.userId,connection.access.grantId,
+        live.id,live.generation,live.agent_remote_id,live.agent_name,this.now()]);
+      return { ...view(live),already_bound:true };
+    });
+  }
+
+  /** Revoke only this grant's reply permit. Native binding/injection remains. */
+  async unbindGrant(connection: McpOAuthConnection, threadId: string): Promise<{ revoked: boolean }> {
+    if (!THREAD.test(threadId)) throw new CapyBindingError(400,"capy_invalid_thread");
+    const actor = connection.access.principal;
+    return this.locked(actor, async tx => {
+      await connection.lock(tx);
+      const result = await tx.query(`UPDATE capy_grant_permits p SET revoked_at=$5,revoke_reason='explicit_unbind'
+        FROM capy_bindings b WHERE p.binding_id=b.id AND p.grant_id=$1 AND p.user_id=$2
+        AND p.organization_id=$3 AND p.workspace_id=$4 AND b.thread_id=$6 AND p.revoked_at IS NULL RETURNING p.id`,
+      [connection.access.grantId,actor.userId,actor.organizationId,actor.workspaceId,this.now(),threadId]);
+      return { revoked:result.rows.length>0 };
+    });
+  }
+
+  private async grantPermit(tx: SqlExecutor, connection: McpOAuthConnection, name: string) {
+    await connection.lock(tx);
+    const actor = connection.access.principal;
+    const permit = (await tx.query<Permit>(`SELECT p.* FROM capy_grant_permits p
+      WHERE p.grant_id=$1 AND p.organization_id=$2 AND p.workspace_id=$3 AND p.user_id=$4
+        AND p.agent_name=$5 AND p.revoked_at IS NULL FOR UPDATE`,
+    [connection.access.grantId,actor.organizationId,actor.workspaceId,actor.userId,name])).rows[0];
+    if (!permit) return null;
+    const row = (await tx.query<Row>("SELECT * FROM capy_bindings WHERE id=$1 FOR UPDATE",[permit.binding_id])).rows[0];
+    if (!row || row.state!=="bound" || row.generation!==permit.binding_generation ||
+      row.agent_remote_id!==permit.agent_remote_id || row.agent_name!==permit.agent_name ||
+      !row.token_ciphertext || capyAgentName(row.thread_id)!==name) return null;
+    return { permit,row:await this.current(tx,actor,row) };
+  }
+
+  /** Send-only OAuth resolver. Never becomes a generic participant/identity. */
+  async agentForGrant(connection: McpOAuthConnection, name: string): Promise<{ name: string; token: string } | null> {
+    if (!name.startsWith("capy-")) return null;
+    const actor = connection.access.principal;
+    try {
+      const selected = await this.locked(actor,tx=>this.grantPermit(tx,connection,name));
+      if (!selected) return null;
+      // Remote verification takes place outside SQL locks. Recheck the same
+      // permit and exact binding/token afterwards, including grant revocation.
+      const found = await this.agentFor(actor.workspaceId,actor.userId,name);
+      if (!found) return null;
+      return this.locked(actor,async tx=>{
+        const current = await this.grantPermit(tx,connection,name);
+        if (!current || current.permit.id!==selected.permit.id || current.row.id!==selected.row.id ||
+          current.row.generation!==selected.row.generation || current.row.agent_remote_id!==selected.row.agent_remote_id ||
+          current.row.token_ciphertext!==selected.row.token_ciphertext ||
+          decryptSecret(current.row.token_ciphertext!,this.deps.keyring)!==found.token) return null;
+        return found;
+      });
+    } catch (error) {
+      if (error instanceof CapyBindingError && error.status!==503) return null;
+      throw error;
+    }
+  }
+
   async retire(actor: CapyActor, id: string): Promise<void> {
     if (!UUID.test(id)) throw new CapyBindingError(404,"capy_binding_not_found");
     await this.locked(actor, async tx => {

src/capy/mcp-permits.ts

@@ -0,0 +1,34 @@
+import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
+import { z } from "zod";
+import type { McpContext } from "../mcp/context.js";
+import { McpOAuthAccessRefused } from "../mcp/oauth/admission.js";
+import { CapyBindingError } from "./bindings.js";
+
+/** Optional OAuth send permits only. No binding creation, transfer or persona pin. */
+export function registerCapyPermitTools(server: McpServer, context: McpContext): void {
+  const run = async (work: () => Promise<unknown>) => {
+    try {
+      const oauth = context.oauth;
+      if (!oauth || !context.capyPermits || oauth.access.principal.userId!==context.principal.userId ||
+        oauth.access.principal.organizationId!==context.workspace.organizationId ||
+        oauth.access.principal.workspaceId!==context.workspace.id ||
+        context.principal.organizationId!==context.workspace.organizationId ||
+        context.principal.workspaceId!==context.workspace.id) throw new McpOAuthAccessRefused();
+      const data = await work();
+      return { content:[{ type:"text" as const,text:JSON.stringify(data) }] };
+    } catch (error) {
+      return { isError:true,content:[{type:"text" as const,text:error instanceof CapyBindingError ? error.code
+        : error instanceof McpOAuthAccessRefused ? error.message : "Capy reply permit is temporarily unavailable."}] };
+    }
+  };
+  const inputSchema = { thread_id:z.string().regex(/^jam_[A-Za-z0-9]{6,64}$/) };
+  const annotations = { readOnlyHint:false,destructiveHint:false,idempotentHint:true,openWorldHint:false };
+  server.registerTool("bind_capy_thread",{
+    description:"Allow this OAuth connection to send replies as an existing Capy thread binding owned by your connected key. Returns the existing binding; creates only a grant-specific send permit. Does not create or transfer a binding, change your approved persona, or grant inbox/channel/history access.",
+    inputSchema,annotations,
+  },({thread_id})=>run(()=>context.capyPermits!.bindGrant(context.oauth!,thread_id)));
+  server.registerTool("unbind_capy_thread",{
+    description:"Revoke only this OAuth connection's Capy reply permit for the selected thread. Native binding custody and inbound delivery remain active; other connections' permits are unchanged.",
+    inputSchema,annotations:{...annotations,destructiveHint:true},
+  },({thread_id})=>run(()=>context.capyPermits!.unbindGrant(context.oauth!,thread_id)));
+}

src/control-tables.ts

@@ -10,6 +10,7 @@ export const CONTROL_TABLES: readonly string[] = [
   "amp_history_syncs",
   "capture_policies",
   "capy_bindings",
+  "capy_grant_permits",
   "cli_credential_refs",
   "coding_agent_credential_blobs",
   "coding_agent_credentials",

src/mcp/capabilities.ts

@@ -24,6 +24,7 @@ export function registerCapabilityDiscovery(server: McpServer, context: McpConte
       amp: Boolean(context.amp),
       factory: Boolean(context.factory),
       launch: Boolean(context.launch),
+      capyPermits: Boolean(context.oauth && context.capyPermits),
       providers: await context.providers(),
     });
     return {

src/mcp/catalog.ts

@@ -95,7 +95,7 @@ export const TOOL_GROUPS: Readonly<Record<ToolGroupId, GroupDefinition>> = {
 export type ToolPlanEntry = { group: ToolGroupId; tools: readonly string[] };
 
 /** The self-host shared-sessions plan, in registration order. */
-export function planToolGroups(options: { amp?: boolean; factory?: boolean; launch?: boolean } = {}): ToolPlanEntry[] {
+export function planToolGroups(options: { amp?: boolean; factory?: boolean; launch?: boolean; capyPermits?: boolean } = {}): ToolPlanEntry[] {
   const groups: ToolGroupId[] = ["shared_session_reads", "team_activity", "relay_messaging"];
   // Launch tools exist only when SELFHOST_FEATURES selects launch; role and
   // connection kind are checked on every call.
@@ -104,6 +104,7 @@ export function planToolGroups(options: { amp?: boolean; factory?: boolean; laun
   if (options.amp) groups.push("amp");
   // Factory tools exist only when SELFHOST_FACTORY_ENABLED is on.
   if (options.factory) groups.push("factory");
+  if (options.capyPermits) groups.push("capy_binding");
   groups.push("capability_discovery");
   return groups.map(group => ({ group, tools: TOOL_GROUPS[group].tools }));
 }
@@ -167,11 +168,13 @@ export function describeCapabilities(input: {
   amp?: boolean;
   factory?: boolean;
   launch?: boolean;
+  capyPermits?: boolean;
   providers: ProviderStatus[];
 }): CapabilityReport {
   const launchRole = input.role === "owner" || input.role === "admin";
   const canLaunch = Boolean(input.launch) && input.authMode !== "oauth" && launchRole;
-  const planned = new Map(planToolGroups({ amp: input.amp, factory: input.factory, launch: input.launch && input.authMode !== "oauth" })
+  const planned = new Map(planToolGroups({ amp: input.amp, factory: input.factory, launch: input.launch && input.authMode !== "oauth",
+    capyPermits: input.capyPermits && input.authMode === "oauth" })
     .map(item => [item.group, item]));
   const groups = (Object.keys(TOOL_GROUPS) as ToolGroupId[]).map((id): CapabilityGroup => {
     const definition = TOOL_GROUPS[id];

src/mcp/context.ts

@@ -65,6 +65,8 @@ export type McpContext = {
   factory?: FactoryMcpDeps;
   /** Internal native send resolver; absent until a complete Capy module is mounted. */
   capy?: Pick<CapyBindings, "agentFor">;
+  /** Explicit exact-grant reply permits; never the native resolver or a persona pin. */
+  capyPermits?: Pick<CapyBindings, "bindGrant" | "unbindGrant" | "agentForGrant">;
   /** Where shared-session read calls are audited (src/mcp/read-audit.ts); absent writes nothing. */
   readAudit?: {
     database: SqlExecutor;

src/mcp/oauth/revocations.ts

@@ -10,6 +10,8 @@ export async function revokeMcpOAuthGrantRows(tx:SqlExecutor,id:string,reason:st
   const grant=(await tx.query<Grant>("SELECT * FROM mcp_oauth_grants WHERE id=$1 FOR UPDATE",[id])).rows[0];
   if(!grant) return false;
   await tx.query(`UPDATE mcp_oauth_grants SET revoked_at=$2,revoked_reason=$3 WHERE id=$1 AND revoked_at IS NULL`,[id,now,reason]);
+  await tx.query(`UPDATE capy_grant_permits SET revoked_at=$2,revoke_reason='grant_revoked'
+    WHERE grant_id=$1 AND revoked_at IS NULL`,[id,now]);
   await revokeMcpOAuthSessionFamily(tx,{
     grantId:id,tokenFamilyId:grant.token_family_id,userId:grant.user_id,
     organizationId:grant.organization_id,workspaceId:grant.workspace_id,