PR #3396
Sections
Review

docs(agentctl): document the ACP session-transition lock so new session paths do not bypass it

main ← feature/serialize-acp-sessio-i7y 1 files +1 −1 PR #3396 ↗

Adds a guardrail note to AGENTS.md that every new session path must hold Adapter.sessionTransitionMu for the full transition, so concurrent WS requests cannot race session/new and session/load.

Why this change

WS requests reach the ACP adapter without serialization. A new path that writes a.sessionID without the transition lock can race NewSession, LoadSession, or ResetSession and corrupt the active session.

What it does

Architecture, end to end

WS dispatch fans out to the adapter without a queue. The mutex serializes the three session transitions that write a.sessionID.

flowchart LR
  WS[WS clients] --> Dispatch[agentctl WS dispatch]
  Dispatch --> Adapter[acp.Adapter]
  Adapter --> Mu[sessionTransitionMu]
  Mu --> New[NewSession]
  Mu --> Load[LoadSession]
  Mu --> Reset[ResetSession + closeSupersededSessionLocked]
  New --> SID[a.sessionID]
  Load --> SID
  Reset --> SID

Key code changes

Drag to pan. Use the + and − buttons to zoom. Click a node to open the full code. The arrows show how the parts interact.

drag to pan · +/− to zoom · click a node for details

Documented transition invariantapps/backend/internal/agentctl/AGENTS.md ↗
Adapter.sessionTransitionMu
Click for details →

Adds the serialization rule so future session paths hold the mutex for the full transition.

Added note
The `acp` transport is split by concern across `adapter_*.go` files: `adapter.go` (core/lifecycle), `adapter_session.go` (initialize/new/load/resume/close), `adapter_prompt.go` (prompt/cancel), `adapter_updates.go` (`session/update` notification fan-out), `adapter_tools.go` (`convertToolCallUpdate` / `convertToolCallResultUpdate` -> normalized payloads), `adapter_permissions.go`, and `adapter_helpers.go`. Agent-specific ACP extensions use the package-private `acpDialect` function table in `dialect.go`; keep observed wire translation in `dialect_<agent>.go`. Dialect hooks return normalized data or request descriptions and never receive `*Adapter` or execute RPCs. Shared capability normalization used by both live sessions and utility probes belongs in `internal/agentctl/acpcompat/`. Tool-call conversion lives in `adapter_tools.go`, not `adapter.go`. See ADR-0043.
The `acp` transport is split by concern across `adapter_*.go` files: `adapter.go` (core/lifecycle), `adapter_session.go` (initialize/new/load/resume/close), `adapter_prompt.go` (prompt/cancel), `adapter_updates.go` (`session/update` notification fan-out), `adapter_tools.go` (`convertToolCallUpdate` / `convertToolCallResultUpdate` -> normalized payloads), `adapter_permissions.go`, and `adapter_helpers.go`. Agent-specific ACP extensions use the package-private `acpDialect` function table in `dialect.go`; keep observed wire translation in `dialect_<agent>.go`. Dialect hooks return normalized data or request descriptions and never receive `*Adapter` or execute RPCs. Shared capability normalization used by both live sessions and utility probes belongs in `internal/agentctl/acpcompat/`. Tool-call conversion lives in `adapter_tools.go`, not `adapter.go`. See ADR-0043. Session lifecycle transitions are serialized: `NewSession`, `LoadSession`, and `ResetSession` (`session/new` plus the `closeSupersededSessionLocked` cleanup) each run under `Adapter.sessionTransitionMu`, because agentctl dispatches WS requests to the adapter without serialization; any new path that writes `a.sessionID` must hold that mutex for the whole transition, not just the write.
type Adapter struct
Click for details →

Defines the lock that the docs now require every session transition to hold.

Adapter fields
type Adapter struct {
    // Session configuration changes are serialized across model and option
    // RPCs. configGeneration is incremented when a change begins so an older
    // completion cannot overwrite a newer selection.
    // Session transitions use a separate mutex because a reset must keep the
    // adapter transitionally consistent from session/new through session/close.
    sessionTransitionMu sync.Mutex
    configChangeMu      sync.Mutex
    configGeneration    uint64
    contextSamples      map[string]contextWindowSample
}
func (a *Adapter) NewSession(ctx context.Context, mcpServers []types.McpServer) (string, error)
Click for details →

Each transition locks the mutex for the full RPC and the superseded-session cleanup.

NewSession / LoadSession / ResetSession
func (a *Adapter) NewSession(ctx context.Context, mcpServers []types.McpServer) (string, error) {
    a.sessionTransitionMu.Lock()
    defer a.sessionTransitionMu.Unlock()
    return a.newSession(ctx, mcpServers)
}

func (a *Adapter) LoadSession(ctx context.Context, sessionID string, mcpServers []types.McpServer) error {
    a.sessionTransitionMu.Lock()
    defer a.sessionTransitionMu.Unlock()
    // ... validates conn, checks LoadSession capability, then calls conn.LoadSession
}

func (a *Adapter) ResetSession(ctx context.Context, mcpServers []types.McpServer) (string, error) {
    a.sessionTransitionMu.Lock()
    defer a.sessionTransitionMu.Unlock()
    previous, conn := a.sessionID, a.acpConn
    newID, err := a.newSession(ctx, mcpServers)
    if err != nil {
        return "", err
    }
    a.closeSupersededSessionLocked(ctx, conn, previous, newID)
    return newID, nil
}
Cleanup helper
// closeSupersededSessionLocked requires sessionTransitionMu to be held so the
// still-current check and session/close request form one transition.
func (a *Adapter) closeSupersededSessionLocked(ctx context.Context, conn *acp.ClientSideConnection, previous, newID string) {
    if previous == "" || previous == newID || conn == nil {
        return
    }
    a.mu.RLock()
    supportsClose := a.capabilities.SessionCapabilities.Close != nil
    stillCurrent := a.sessionID == newID
    a.mu.RUnlock()
    if !supportsClose || !stillCurrent {
        return
    }
    closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), closeSupersededSessionTimeout)
    defer cancel()
    if _, err := conn.CloseSession(closeCtx, acp.CloseSessionRequest{
        SessionId: acp.SessionId(previous),
    }); err != nil {
        a.logger.Warn("failed to close superseded session after reset",
            zap.String("session_id", previous), zap.Error(err))
    }
}
Read the changes as a list

Documented transition invariant

apps/backend/internal/agentctl/AGENTS.md

Adds the serialization rule so future session paths hold the mutex for the full transition.

Added note
The `acp` transport is split by concern across `adapter_*.go` files: `adapter.go` (core/lifecycle), `adapter_session.go` (initialize/new/load/resume/close), `adapter_prompt.go` (prompt/cancel), `adapter_updates.go` (`session/update` notification fan-out), `adapter_tools.go` (`convertToolCallUpdate` / `convertToolCallResultUpdate` -> normalized payloads), `adapter_permissions.go`, and `adapter_helpers.go`. Agent-specific ACP extensions use the package-private `acpDialect` function table in `dialect.go`; keep observed wire translation in `dialect_<agent>.go`. Dialect hooks return normalized data or request descriptions and never receive `*Adapter` or execute RPCs. Shared capability normalization used by both live sessions and utility probes belongs in `internal/agentctl/acpcompat/`. Tool-call conversion lives in `adapter_tools.go`, not `adapter.go`. See ADR-0043.
The `acp` transport is split by concern across `adapter_*.go` files: `adapter.go` (core/lifecycle), `adapter_session.go` (initialize/new/load/resume/close), `adapter_prompt.go` (prompt/cancel), `adapter_updates.go` (`session/update` notification fan-out), `adapter_tools.go` (`convertToolCallUpdate` / `convertToolCallResultUpdate` -> normalized payloads), `adapter_permissions.go`, and `adapter_helpers.go`. Agent-specific ACP extensions use the package-private `acpDialect` function table in `dialect.go`; keep observed wire translation in `dialect_<agent>.go`. Dialect hooks return normalized data or request descriptions and never receive `*Adapter` or execute RPCs. Shared capability normalization used by both live sessions and utility probes belongs in `internal/agentctl/acpcompat/`. Tool-call conversion lives in `adapter_tools.go`, not `adapter.go`. See ADR-0043. Session lifecycle transitions are serialized: `NewSession`, `LoadSession`, and `ResetSession` (`session/new` plus the `closeSupersededSessionLocked` cleanup) each run under `Adapter.sessionTransitionMu`, because agentctl dispatches WS requests to the adapter without serialization; any new path that writes `a.sessionID` must hold that mutex for the whole transition, not just the write.

Mutex that serializes transitions

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go

Defines the lock that the docs now require every session transition to hold.

Adapter fields
type Adapter struct {
    // Session configuration changes are serialized across model and option
    // RPCs. configGeneration is incremented when a change begins so an older
    // completion cannot overwrite a newer selection.
    // Session transitions use a separate mutex because a reset must keep the
    // adapter transitionally consistent from session/new through session/close.
    sessionTransitionMu sync.Mutex
    configChangeMu      sync.Mutex
    configGeneration    uint64
    contextSamples      map[string]contextWindowSample
}

Three locked entry points

apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go

Each transition locks the mutex for the full RPC and the superseded-session cleanup.

NewSession / LoadSession / ResetSession
func (a *Adapter) NewSession(ctx context.Context, mcpServers []types.McpServer) (string, error) {
    a.sessionTransitionMu.Lock()
    defer a.sessionTransitionMu.Unlock()
    return a.newSession(ctx, mcpServers)
}

func (a *Adapter) LoadSession(ctx context.Context, sessionID string, mcpServers []types.McpServer) error {
    a.sessionTransitionMu.Lock()
    defer a.sessionTransitionMu.Unlock()
    // ... validates conn, checks LoadSession capability, then calls conn.LoadSession
}

func (a *Adapter) ResetSession(ctx context.Context, mcpServers []types.McpServer) (string, error) {
    a.sessionTransitionMu.Lock()
    defer a.sessionTransitionMu.Unlock()
    previous, conn := a.sessionID, a.acpConn
    newID, err := a.newSession(ctx, mcpServers)
    if err != nil {
        return "", err
    }
    a.closeSupersededSessionLocked(ctx, conn, previous, newID)
    return newID, nil
}
Cleanup helper
// closeSupersededSessionLocked requires sessionTransitionMu to be held so the
// still-current check and session/close request form one transition.
func (a *Adapter) closeSupersededSessionLocked(ctx context.Context, conn *acp.ClientSideConnection, previous, newID string) {
    if previous == "" || previous == newID || conn == nil {
        return
    }
    a.mu.RLock()
    supportsClose := a.capabilities.SessionCapabilities.Close != nil
    stillCurrent := a.sessionID == newID
    a.mu.RUnlock()
    if !supportsClose || !stillCurrent {
        return
    }
    closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), closeSupersededSessionTimeout)
    defer cancel()
    if _, err := conn.CloseSession(closeCtx, acp.CloseSessionRequest{
        SessionId: acp.SessionId(previous),
    }); err != nil {
        a.logger.Warn("failed to close superseded session after reset",
            zap.String("session_id", previous), zap.Error(err))
    }
}

Risk

1 / 10 Low
1 low5 medium10 high

Why this score

  • Docs-only change to AGENTS.md, no runtime code or behavior change.
  • No data, API, or config change and no migration.
  • Revert is a single-line docs revert with no side effects.

Trade-offs and review notes

Where to look first

  1. Confirm the added sentence matches the actual lock scope in adapter_session.go: NewSession, LoadSession, ResetSession, and closeSupersededSessionLocked.
  2. Check wording is precise about holding the mutex for the whole transition, not just the write to a.sessionID.
  3. Verify no other sessionID writers exist outside the three listed paths.