feat(acp): delete sessions after draining owned work

Assisted-By: docker-agent
Signed-off-by: David Gageot <david.gageot@docker.com>

docs/features/acp/index.md

@@ -170,6 +170,16 @@ Concurrent close requests join the same cleanup. Canceling a close request only
 
 Server shutdown rejects new work, cancels admitted initialization/session/list operations, and drains them before closing the session store. Shutdown is a final join rather than a bounded timeout: an uncooperative runtime or tool can delay it. Toolset stop errors are surfaced, not treated as successful cleanup. These guarantees do not add disposal support to toolsets whose resources fall outside the existing lifecycle contract, nor undo already-issued client I/O.
 
+## Deleting Sessions
+
+`session/delete` permanently removes a root conversation's stored history, descendant sessions, and stored generated-media records/blobs. It does not remove files from the workspace. The agent advertises `sessionCapabilities.delete`; the pinned Go SDK still names the handler `UnstableDeleteSession`, but the wire method is `session/delete`. An empty session ID is invalid; already-deleted and unknown IDs succeed without creating a session. Subsequent load/resume requests for deleted history return not found.
+
+Deletion closes the root session first, canceling its foreground work and joining loads/resumes, runtime/background tasks, and client MCP cleanup before deleting storage. Cleanup failure preserves history and remains an error for that lifecycle. An ordinary store error leaves the session closed and can be retried; SQLite subtree/media changes occur in one transaction.
+
+Direct deletion of child sessions is rejected. Independently loaded descendants must be closed, with successful cleanup, before their root can be deleted. While a delete is admitted, new session construction and reconnects are temporarily rejected agent-wide to prevent a descendant from being loaded during the cascade. A delete also returns busy if another session already has an in-flight constructor whose ancestry is not yet known. Different-ID deletes are serialized by returning busy; concurrent requests for the same ID join one operation. Unrelated already-running sessions are not canceled.
+
+Canceling a joining request stops only that caller's wait. Canceling the initiating request before storage deletion lets cleanup finish but skips deletion, keeping the admission barrier until the drain ends. Cancellation during a database call does not prove whether it committed; retrying deletion is safe. Shutdown joins any admitted deletion before the caller closes the store. These guarantees cover this agent instance, not other processes concurrently writing the same database; they do not provide an atomic filesystem revocation or a deadline for uncooperative cleanup.
+
 ## Plan Snapshots
 
 Todo tool results with typed todo metadata produce complete ACP plan snapshots. An empty todo snapshot sends `entries: []`, replacing and clearing the previous plan; completed entries remain visible until the todo storage is cleared. Missing or unrelated metadata does not change the plan, and textual tool output is not parsed to infer one. This uses the stable v1 `plan` update, not ID-based plan-removal extensions.

pkg/acp/agent.go

@@ -49,6 +49,7 @@ type Agent struct {
 	stopErr      error
 	cleanupErr   error
 	owned        map[*Session]struct{}
+	deletion     *sessionDeletion
 }
 
 var _ acp.Agent = (*Agent)(nil)
@@ -312,6 +313,7 @@ func (a *Agent) Initialize(ctx context.Context, params acp.InitializeRequest) (a
 			SessionCapabilities: acp.SessionCapabilities{
 				AdditionalDirectories: &acp.SessionAdditionalDirectoriesCapabilities{},
 				Close:                 &acp.SessionCloseCapabilities{},
+				Delete:                &acp.SessionDeleteCapabilities{},
 				List:                  &acp.SessionListCapabilities{},
 				Resume:                &acp.SessionResumeCapabilities{},
 			},
@@ -426,6 +428,9 @@ func (a *Agent) NewSession(ctx context.Context, params acp.NewSessionRequest) (_
 	if err != nil {
 		return acp.NewSessionResponse{}, err
 	}
+	a.mu.Lock()
+	op.lifecycle.newRoot = true
+	a.mu.Unlock()
 	defer a.finishOperation(op)
 	acpSess, defaultAgent, err := a.newRuntime(ctx, workingDir, servers)
 	stored := false

pkg/acp/delete.go

@@ -0,0 +1,136 @@
+package acp
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/coder/acp-go-sdk"
+
+	"github.com/docker/docker-agent/pkg/session"
+)
+
+type sessionDeletion struct {
+	id   string
+	done chan struct{}
+	err  error
+}
+
+// UnstableDeleteSession implements the SDK's session/delete handler.
+func (a *Agent) UnstableDeleteSession(ctx context.Context, params acp.UnstableDeleteSessionRequest) (acp.UnstableDeleteSessionResponse, error) {
+	sid := string(params.SessionId)
+	if sid == "" {
+		return acp.UnstableDeleteSessionResponse{}, acp.NewInvalidParams("sessionId is required")
+	}
+	a.mu.Lock()
+	if err := ctx.Err(); err != nil {
+		a.mu.Unlock()
+		return acp.UnstableDeleteSessionResponse{}, err
+	}
+	if a.stopped || a.team == nil {
+		a.mu.Unlock()
+		return acp.UnstableDeleteSessionResponse{}, acp.NewInvalidRequest("agent is stopped or not initialized")
+	}
+	deletion := a.deletion
+	if deletion != nil {
+		a.mu.Unlock()
+		if deletion.id != sid {
+			return acp.UnstableDeleteSessionResponse{}, acp.NewInvalidRequest("another session deletion is in progress")
+		}
+	} else {
+		// An unrelated constructor could be loading a descendant not yet registered.
+		for op := range a.pending {
+			if op.lifecycle != nil && op.lifecycle.id != sid {
+				a.mu.Unlock()
+				return acp.UnstableDeleteSessionResponse{}, acp.NewInvalidRequest("session construction in progress; retry deletion after it finishes")
+			}
+		}
+		deletion = &sessionDeletion{id: sid, done: make(chan struct{})}
+		a.deletion = deletion
+		deleteCtx, op := a.startOperationLocked(ctx, nil)
+		a.mu.Unlock()
+		go func() {
+			defer a.finishOperation(op)
+			err := a.deleteSession(deleteCtx, sid)
+			a.mu.Lock()
+			defer a.mu.Unlock()
+			deletion.err = err
+			a.deletion = nil
+			close(deletion.done)
+		}()
+	}
+	if err := waitForCleanup(ctx, deletion.done); err != nil {
+		return acp.UnstableDeleteSessionResponse{}, err
+	}
+	return acp.UnstableDeleteSessionResponse{}, deletion.err
+}
+
+func (a *Agent) deleteSession(ctx context.Context, sid string) error {
+	saved, err := a.sessionStore.GetSession(ctx, sid)
+	if err != nil && !errors.Is(err, session.ErrNotFound) {
+		return fmt.Errorf("reading session before deletion: %w", err)
+	}
+	if saved != nil && saved.ParentID != "" {
+		return acp.NewInvalidParams("delete the root session, not an individual child session")
+	}
+
+	a.mu.Lock()
+	candidates := make(map[string]struct{})
+	for owned := range a.owned {
+		if owned.id != sid {
+			candidates[owned.id] = struct{}{}
+		}
+	}
+	for id, lifecycle := range a.lifecycles {
+		if id != sid && lifecycle.err != nil && !lifecycle.newRoot {
+			candidates[id] = struct{}{}
+		}
+	}
+	a.mu.Unlock()
+	for candidate := range candidates {
+		descendant, err := a.isStoredDescendant(ctx, candidate, sid)
+		if err != nil {
+			return err
+		}
+		if descendant {
+			return acp.NewInvalidRequest("a descendant session is loaded or has unresolved cleanup; close it before deleting the root")
+		}
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	a.mu.Lock()
+	lifecycle := a.closeSessionLocked(context.WithoutCancel(ctx), sid)
+	a.mu.Unlock()
+	// Keep admission blocked even if the caller stops waiting for the drain.
+	<-lifecycle.done
+	if lifecycle.err != nil {
+		return lifecycle.err
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	if err := a.sessionStore.DeleteSession(ctx, sid); err != nil && !errors.Is(err, session.ErrNotFound) {
+		return fmt.Errorf("deleting session: %w", err)
+	}
+	return nil
+}
+
+func (a *Agent) isStoredDescendant(ctx context.Context, id, root string) (bool, error) {
+	seen := make(map[string]struct{})
+	for id != "" {
+		if id == root {
+			return true, nil
+		}
+		if _, ok := seen[id]; ok {
+			return false, errors.New("cyclic session ancestry")
+		}
+		seen[id] = struct{}{}
+		saved, err := a.sessionStore.GetSession(ctx, id)
+		if err != nil {
+			return false, fmt.Errorf("checking session ancestry: %w", err)
+		}
+		id = saved.ParentID
+	}
+	return false, nil
+}

pkg/acp/delete_test.go

@@ -0,0 +1,486 @@
+package acp
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"sync/atomic"
+	"testing"
+	"testing/synctest"
+	"time"
+
+	acpsdk "github.com/coder/acp-go-sdk"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/docker/docker-agent/pkg/agent"
+	"github.com/docker/docker-agent/pkg/config"
+	"github.com/docker/docker-agent/pkg/modelsdev"
+	"github.com/docker/docker-agent/pkg/session"
+	"github.com/docker/docker-agent/pkg/session/sqlitestore"
+	"github.com/docker/docker-agent/pkg/team"
+	"github.com/docker/docker-agent/pkg/teamloader"
+	agenttool "github.com/docker/docker-agent/pkg/tools/builtin/agent"
+)
+
+func deleteAsync(a *Agent, ctx context.Context, sid string) <-chan error {
+	done := make(chan error, 1)
+	go func() {
+		_, err := a.UnstableDeleteSession(ctx, acpsdk.UnstableDeleteSessionRequest{SessionId: acpsdk.SessionId(sid)})
+		done <- err
+	}()
+	return done
+}
+
+type deletionStore struct {
+	session.Store
+
+	started, release chan struct{}
+	calls            atomic.Int32
+	failure          error
+}
+
+func (s *deletionStore) DeleteSession(ctx context.Context, id string) error {
+	s.calls.Add(1)
+	if s.started != nil {
+		close(s.started)
+		<-s.release
+	}
+	if s.failure != nil {
+		return s.failure
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	return s.Store.DeleteSession(ctx, id)
+}
+
+func TestDeleteSessionSQLiteTreeAndIdempotency(t *testing.T) {
+	t.Parallel()
+	store, err := sqlitestore.New(t.Context(), filepath.Join(t.TempDir(), "sessions.db"))
+	require.NoError(t, err)
+	defer func() { require.NoError(t, store.Close()) }()
+	wd := t.TempDir()
+	file := filepath.Join(wd, "keep.txt")
+	require.NoError(t, os.WriteFile(file, []byte("workspace"), 0o600))
+	root := session.New(session.WithWorkingDir(wd), session.WithUserMessage("root history"))
+	require.NoError(t, store.AddSession(t.Context(), root))
+	child := session.New(session.WithParentID(root.ID), session.WithWorkingDir(wd))
+	require.NoError(t, store.AddSubSession(t.Context(), root.ID, child))
+	a := NewAgent(nil, &config.RuntimeConfig{}, store)
+	a.team = team.New()
+	a.loadTeam = func(context.Context, string) (*teamloader.LoadResult, error) {
+		loaded, _ := loadedLifecycleTeam(t)
+		return loaded, nil
+	}
+	defer func() { require.NoError(t, a.Stop(t.Context())) }()
+	out := &captureWriter{}
+	captureReplay(t, a, out)
+	_, err = a.ResumeSession(t.Context(), acpsdk.ResumeSessionRequest{SessionId: acpsdk.SessionId(root.ID), Cwd: wd})
+	require.NoError(t, err)
+	for range 2 {
+		_, err = a.UnstableDeleteSession(t.Context(), acpsdk.UnstableDeleteSessionRequest{SessionId: acpsdk.SessionId(root.ID)})
+		require.NoError(t, err)
+	}
+	for _, id := range []string{root.ID, child.ID} {
+		_, err = store.GetSession(t.Context(), id)
+		require.ErrorIs(t, err, session.ErrNotFound)
+	}
+	listed, err := a.ListSessions(t.Context(), acpsdk.ListSessionsRequest{})
+	require.NoError(t, err)
+	assert.Empty(t, listed.Sessions)
+	_, err = a.LoadSession(t.Context(), acpsdk.LoadSessionRequest{SessionId: acpsdk.SessionId(root.ID), Cwd: wd})
+	require.Error(t, err)
+	_, err = a.ResumeSession(t.Context(), acpsdk.ResumeSessionRequest{SessionId: acpsdk.SessionId(root.ID), Cwd: wd})
+	require.Error(t, err)
+	require.FileExists(t, file)
+}
+
+func TestDeleteWaitsForDrainAndBlocksReconnectionThroughStoreCall(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		rt := &barrierRuntime{started: make(chan struct{}), release: make(chan struct{})}
+		s.rt = rt
+		store := &deletionStore{Store: a.sessionStore, started: make(chan struct{}), release: make(chan struct{})}
+		a.sessionStore = store
+		deleted := deleteAsync(a, t.Context(), s.id)
+		<-rt.started
+		synctest.Wait()
+		assertPending(t, deleted)
+		assert.Zero(t, store.calls.Load())
+		close(rt.release)
+		<-store.started
+		_, err := a.LoadSession(t.Context(), loadRequest(s))
+		require.ErrorContains(t, err, "deletion in progress")
+		_, err = a.ResumeSession(t.Context(), acpsdk.ResumeSessionRequest{SessionId: acpsdk.SessionId(s.id), Cwd: s.workingDir})
+		require.ErrorContains(t, err, "deletion in progress")
+		_, err = a.NewSession(t.Context(), acpsdk.NewSessionRequest{Cwd: s.workingDir})
+		require.ErrorContains(t, err, "deletion in progress")
+		joined := deleteAsync(a, t.Context(), s.id)
+		synctest.Wait()
+		assertPending(t, joined)
+		_, err = a.UnstableDeleteSession(t.Context(), acpsdk.UnstableDeleteSessionRequest{SessionId: "other"})
+		require.ErrorContains(t, err, "another session deletion")
+		close(store.release)
+		require.NoError(t, <-deleted)
+		require.NoError(t, <-joined)
+		assert.Equal(t, int32(1), store.calls.Load())
+		_, err = store.GetSession(t.Context(), s.id)
+		require.ErrorIs(t, err, session.ErrNotFound)
+	})
+}
+
+func TestDeleteCancelsPromptAndDrainsBeforeDeletion(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		rt := &drainingPromptRuntime{started: make(chan struct{}), release: make(chan struct{}), stopped: make(chan struct{})}
+		s.rt = rt
+		out := &captureWriter{}
+		captureReplay(t, a, out)
+		store := &deletionStore{Store: a.sessionStore}
+		a.sessionStore = store
+		prompted := make(chan error, 1)
+		go func() {
+			_, err := a.Prompt(t.Context(), acpsdk.PromptRequest{SessionId: acpsdk.SessionId(s.id), Prompt: []acpsdk.ContentBlock{acpsdk.TextBlock("work")}})
+			prompted <- err
+		}()
+		<-rt.started
+		deleted := deleteAsync(a, t.Context(), s.id)
+		synctest.Wait()
+		assertPending(t, deleted)
+		assert.Zero(t, store.calls.Load())
+		close(rt.release)
+		require.NoError(t, <-prompted)
+		require.NoError(t, <-deleted)
+		assert.Equal(t, int32(1), store.calls.Load())
+	})
+}
+
+func TestDeleteCanceledOwnerDrainsButRetainsHistory(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		rt := &barrierRuntime{started: make(chan struct{}), release: make(chan struct{})}
+		s.rt = rt
+		store := &deletionStore{Store: a.sessionStore}
+		a.sessionStore = store
+		ctx, cancel := context.WithCancel(t.Context())
+		defer cancel()
+		deleted := deleteAsync(a, ctx, s.id)
+		<-rt.started
+		cancel()
+		require.ErrorIs(t, <-deleted, context.Canceled)
+		_, err := a.ResumeSession(t.Context(), acpsdk.ResumeSessionRequest{SessionId: acpsdk.SessionId(s.id), Cwd: s.workingDir})
+		require.ErrorContains(t, err, "deletion in progress")
+		close(rt.release)
+		synctest.Wait()
+		assert.Zero(t, store.calls.Load())
+		_, err = store.GetSession(t.Context(), s.id)
+		require.NoError(t, err)
+		require.NoError(t, <-deleteAsync(a, t.Context(), s.id))
+	})
+}
+
+func TestDeleteWaiterCancellationDoesNotCancelOwner(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		store := &deletionStore{Store: a.sessionStore, started: make(chan struct{}), release: make(chan struct{})}
+		a.sessionStore = store
+		owner := deleteAsync(a, t.Context(), s.id)
+		<-store.started
+		ctx, cancel := context.WithCancel(t.Context())
+		waiter := deleteAsync(a, ctx, s.id)
+		synctest.Wait()
+		cancel()
+		require.ErrorIs(t, <-waiter, context.Canceled)
+		close(store.release)
+		require.NoError(t, <-owner)
+		_, err := store.GetSession(t.Context(), s.id)
+		require.ErrorIs(t, err, session.ErrNotFound)
+	})
+}
+
+func TestStopJoinsDeletionStoreCall(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		store := &deletionStore{Store: a.sessionStore, started: make(chan struct{}), release: make(chan struct{})}
+		a.sessionStore = store
+		deleted := deleteAsync(a, t.Context(), s.id)
+		<-store.started
+		stopped := make(chan error, 1)
+		go func() { stopped <- a.Stop(t.Context()) }()
+		synctest.Wait()
+		assertPending(t, stopped)
+		close(store.release)
+		require.ErrorIs(t, <-deleted, context.Canceled)
+		require.NoError(t, <-stopped)
+		_, err := store.GetSession(t.Context(), s.id)
+		require.NoError(t, err)
+	})
+}
+
+func TestDeleteFailuresPreserveHistory(t *testing.T) {
+	t.Parallel()
+	for _, cleanup := range []bool{false, true} {
+		t.Run(map[bool]string{false: "store", true: "cleanup"}[cleanup], func(t *testing.T) {
+			persisted := session.New(session.WithWorkingDir(t.TempDir()))
+			a := NewAgent(nil, nil, session.NewInMemorySessionStore())
+			a.team = team.New()
+			require.NoError(t, a.sessionStore.AddSession(t.Context(), persisted))
+			s := &Session{id: persisted.ID, sess: persisted, rt: &fakeRuntime{}}
+			_, registered, err := registerTestSession(t.Context(), a, s)
+			require.NoError(t, err)
+			require.True(t, registered)
+			defer func() {
+				err := a.Stop(t.Context())
+				if cleanup {
+					require.ErrorContains(t, err, "cleanup failed")
+				} else {
+					require.NoError(t, err)
+				}
+			}()
+			store := &deletionStore{Store: a.sessionStore}
+			a.sessionStore = store
+			if cleanup {
+				rt := &barrierRuntime{started: make(chan struct{}), release: make(chan struct{}), err: errors.New("cleanup failed")}
+				close(rt.release)
+				s.rt = rt
+			} else {
+				store.failure = errors.New("store failed")
+			}
+			_, err = a.UnstableDeleteSession(t.Context(), acpsdk.UnstableDeleteSessionRequest{SessionId: acpsdk.SessionId(s.id)})
+			require.Error(t, err)
+			_, err = store.GetSession(t.Context(), s.id)
+			require.NoError(t, err)
+			if cleanup {
+				assert.Zero(t, store.calls.Load())
+				require.ErrorContains(t, <-deleteAsync(a, t.Context(), s.id), "cleanup failed")
+			} else {
+				store.failure = nil
+				require.NoError(t, <-deleteAsync(a, t.Context(), s.id))
+			}
+		})
+	}
+}
+
+func TestDeleteRejectsLoadedDescendantsAndDirectChildren(t *testing.T) {
+	t.Parallel()
+	a, root := newResumeFixture(t, t.TempDir())
+	child := session.New(session.WithParentID(root.id), session.WithWorkingDir(root.workingDir))
+	grandchild := session.New(session.WithParentID(child.ID), session.WithWorkingDir(root.workingDir))
+	require.NoError(t, a.sessionStore.AddSession(t.Context(), child))
+	require.NoError(t, a.sessionStore.AddSession(t.Context(), grandchild))
+	loaded := &Session{id: grandchild.ID, sess: grandchild, workingDir: root.workingDir, rt: &fakeRuntime{}}
+	_, ok, err := registerTestSession(t.Context(), a, loaded)
+	require.NoError(t, err)
+	require.True(t, ok)
+	require.ErrorContains(t, <-deleteAsync(a, t.Context(), root.id), "descendant")
+	require.ErrorContains(t, <-deleteAsync(a, t.Context(), child.ID), "root session")
+	assert.Same(t, root, a.sessions[root.id])
+	_, err = a.CloseSession(t.Context(), acpsdk.CloseSessionRequest{SessionId: acpsdk.SessionId(grandchild.ID)})
+	require.NoError(t, err)
+	require.NoError(t, <-deleteAsync(a, t.Context(), root.id))
+}
+
+func TestDeleteCancelsSameSessionConstructor(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		_, err := a.CloseSession(t.Context(), acpsdk.CloseSessionRequest{SessionId: acpsdk.SessionId(s.id)})
+		require.NoError(t, err)
+		entered, release := make(chan struct{}), make(chan struct{})
+		loaded, ts := loadedLifecycleTeam(t)
+		a.loadTeam = func(ctx context.Context, _ string) (*teamloader.LoadResult, error) {
+			close(entered)
+			<-release
+			return loaded, nil
+		}
+		resumed := make(chan error, 1)
+		go func() {
+			_, err := a.ResumeSession(t.Context(), acpsdk.ResumeSessionRequest{SessionId: acpsdk.SessionId(s.id), Cwd: s.workingDir})
+			resumed <- err
+		}()
+		<-entered
+		deleted := deleteAsync(a, t.Context(), s.id)
+		synctest.Wait()
+		assertPending(t, deleted)
+		close(release)
+		require.Error(t, <-resumed)
+		require.NoError(t, <-deleted)
+		assert.Equal(t, int32(1), ts.stops.Load())
+		assert.Empty(t, a.sessions)
+	})
+}
+
+func TestDeleteRejectsUnknownConcurrentConstructor(t *testing.T) {
+	t.Parallel()
+	a, s := newResumeFixture(t, t.TempDir())
+	_, op, err := a.beginSessionConstruction(t.Context(), "possible-child")
+	require.NoError(t, err)
+	_, err = a.UnstableDeleteSession(t.Context(), acpsdk.UnstableDeleteSessionRequest{SessionId: acpsdk.SessionId(s.id)})
+	require.ErrorContains(t, err, "construction in progress")
+	a.finishOperation(op)
+	require.NoError(t, <-deleteAsync(a, t.Context(), s.id))
+}
+
+func TestDeleteWireAndCapability(t *testing.T) {
+	t.Parallel()
+	a := NewAgent(config.NewBytesSource("test.yaml", nil), &config.RuntimeConfig{}, session.NewInMemorySessionStore())
+	a.loadTeam = func(context.Context, string) (*teamloader.LoadResult, error) {
+		loaded, _ := loadedLifecycleTeam(t)
+		return loaded, nil
+	}
+	defer func() { require.NoError(t, a.Stop(t.Context())) }()
+	response, err := a.Initialize(t.Context(), acpsdk.InitializeRequest{})
+	require.NoError(t, err)
+	assert.NotNil(t, response.AgentCapabilities.SessionCapabilities.Delete)
+	input, send := io.Pipe()
+	receive, output := io.Pipe()
+	conn := acpsdk.NewAgentSideConnection(a, output, input)
+	t.Cleanup(func() { _ = send.Close(); _ = receive.Close(); <-conn.Done() })
+	encoder, decoder := json.NewEncoder(send), json.NewDecoder(receive)
+	for i, id := range []string{"", "missing", "missing"} {
+		require.NoError(t, encoder.Encode(map[string]any{"jsonrpc": "2.0", "id": i, "method": "session/delete", "params": map[string]any{"sessionId": id}}))
+		var result struct {
+			ID     int                  `json:"id"`
+			Result json.RawMessage      `json:"result"`
+			Error  *acpsdk.RequestError `json:"error"`
+		}
+		require.NoError(t, decoder.Decode(&result))
+		assert.Equal(t, i, result.ID)
+		if id == "" {
+			require.NotNil(t, result.Error)
+			assert.Equal(t, -32602, result.Error.Code)
+		} else {
+			require.Nil(t, result.Error)
+			assert.JSONEq(t, `{}`, string(result.Result))
+		}
+	}
+}
+
+func TestDeleteIgnoresFailedUnpersistedNewRoot(t *testing.T) {
+	t.Parallel()
+	a := NewAgent(nil, nil, failingAddStore{session.NewInMemorySessionStore()})
+	a.team = team.New()
+	wd := t.TempDir()
+	saved := session.New(session.WithWorkingDir(wd))
+	require.NoError(t, a.sessionStore.(failingAddStore).Store.AddSession(t.Context(), saved))
+	ts := &barrierToolset{started: make(chan struct{}), release: make(chan struct{}), err: errors.New("cleanup failed")}
+	close(ts.release)
+	root := agent.New("root", "test", agent.WithToolSets(ts), agent.WithModel(&mockProvider{id: modelsdev.NewID("test", "delete")}))
+	a.loadTeam = func(context.Context, string) (*teamloader.LoadResult, error) {
+		return &teamloader.LoadResult{Team: team.New(team.WithAgents(root))}, nil
+	}
+	for _, started := range root.StartToolSets(t.Context(), 0) {
+		require.NoError(t, (<-started).Err)
+	}
+	_, err := a.NewSession(t.Context(), acpsdk.NewSessionRequest{Cwd: wd})
+	require.ErrorContains(t, err, "store failed")
+	require.ErrorContains(t, err, "cleanup failed")
+	require.NoError(t, <-deleteAsync(a, t.Context(), saved.ID))
+	require.NoError(t, <-deleteAsync(a, t.Context(), "never existed"))
+	require.ErrorContains(t, a.Stop(t.Context()), "cleanup failed", "original cleanup failure remains sticky")
+}
+
+func TestDeleteJoinsLoadReplay(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		s.sess.AddMessage(session.UserMessage("history"))
+		entered, release := make(chan struct{}), make(chan struct{})
+		out := &captureWriter{failOn: func(n int) error {
+			if n == 1 {
+				close(entered)
+				<-release
+			}
+			return nil
+		}}
+		captureReplay(t, a, out)
+		store := &deletionStore{Store: a.sessionStore}
+		a.sessionStore = store
+		loaded := make(chan error, 1)
+		go func() { _, err := a.LoadSession(t.Context(), loadRequest(s)); loaded <- err }()
+		<-entered
+		deleted := deleteAsync(a, t.Context(), s.id)
+		synctest.Wait()
+		assertPending(t, deleted)
+		assert.Zero(t, store.calls.Load())
+		close(release)
+		require.Error(t, <-loaded)
+		require.NoError(t, <-deleted)
+		assert.Equal(t, int32(1), store.calls.Load())
+	})
+}
+
+func TestDeleteRejectsClosingDescendant(t *testing.T) {
+	t.Parallel()
+	synctest.Test(t, func(t *testing.T) {
+		a, s := newResumeFixture(t, t.TempDir())
+		child := session.New(session.WithParentID(s.id), session.WithWorkingDir(s.workingDir))
+		require.NoError(t, a.sessionStore.AddSession(t.Context(), child))
+		rt := &barrierRuntime{started: make(chan struct{}), release: make(chan struct{})}
+		owned := &Session{id: child.ID, sess: child, workingDir: s.workingDir, rt: rt}
+		_, ok, err := registerTestSession(t.Context(), a, owned)
+		require.NoError(t, err)
+		require.True(t, ok)
+		closed := closeSessionAsync(a, t.Context(), child.ID)
+		<-rt.started
+		require.ErrorContains(t, <-deleteAsync(a, t.Context(), s.id), "descendant")
+		close(rt.release)
+		require.NoError(t, <-closed)
+		require.NoError(t, <-deleteAsync(a, t.Context(), s.id))
+	})
+}
+
+func TestDeleteStopsClientMCPBackgroundWork(t *testing.T) {
+	t.Parallel()
+	a := clientMCPAgent(t)
+	a.loadTeam = func(context.Context, string) (*teamloader.LoadResult, error) {
+		worker := agent.New("worker", "test", agent.WithModel(&clientMCPWaitProvider{mockProvider{id: modelsdev.NewID("test", "worker")}}))
+		root := agent.New("root", "test", agent.WithModel(&clientMCPBackgroundProvider{mockProvider: mockProvider{id: modelsdev.NewID("test", "root")}}), agent.WithSubAgents(worker), agent.WithToolSets(agenttool.New()))
+		return &teamloader.LoadResult{Team: team.New(team.WithAgents(root, worker))}, nil
+	}
+	markers, wd := t.TempDir(), t.TempDir()
+	store, err := sqlitestore.New(t.Context(), filepath.Join(t.TempDir(), "sessions.db"))
+	require.NoError(t, err)
+	defer func() { require.NoError(t, store.Close()) }()
+	a.sessionStore = store
+	defer func() { require.NoError(t, a.Stop(t.Context())) }()
+	out := &captureWriter{}
+	captureReplay(t, a, out)
+	spec := clientServerSpec(t, "background", "value")
+	spec.Stdio.Env = append(spec.Stdio.Env, acpsdk.EnvVariable{Name: "ACP_CALL_STARTED", Value: filepath.Join(markers, "called")}, acpsdk.EnvVariable{Name: "ACP_STOPPED", Value: filepath.Join(markers, "stopped")})
+	created, err := a.NewSession(t.Context(), acpsdk.NewSessionRequest{Cwd: wd, McpServers: []acpsdk.McpServer{spec}})
+	require.NoError(t, err)
+	rootSession := a.sessions[string(created.SessionId)].sess
+	rootSession.SetToolsApproved(true)
+	ctx, cancel := context.WithTimeout(t.Context(), 15*time.Second)
+	defer cancel()
+	_, err = a.Prompt(ctx, acpsdk.PromptRequest{SessionId: created.SessionId, Prompt: []acpsdk.ContentBlock{acpsdk.TextBlock("dispatch")}})
+	require.NoError(t, err)
+	require.Eventually(t, func() bool { _, err := os.Stat(filepath.Join(markers, "called")); return err == nil }, 5*time.Second, time.Millisecond)
+	_, err = a.UnstableDeleteSession(ctx, acpsdk.UnstableDeleteSessionRequest{SessionId: created.SessionId})
+	require.NoError(t, err)
+	assert.FileExists(t, filepath.Join(markers, "stopped"))
+	all, err := store.GetSessions(t.Context())
+	require.NoError(t, err)
+	assert.Empty(t, all)
+	var childIDs []string
+	for _, item := range rootSession.MessagesSnapshot() {
+		if item.SubSession != nil {
+			childIDs = append(childIDs, item.SubSession.ID)
+		}
+	}
+	require.NotEmpty(t, childIDs, "background child must finish attaching before delete returns")
+	for _, id := range childIDs {
+		_, err := store.GetSession(t.Context(), id)
+		require.ErrorIs(t, err, session.ErrNotFound)
+	}
+}

pkg/acp/lifecycle.go

@@ -13,6 +13,7 @@ import (
 // sessionLifecycle is also the publication token for constructors admitted before close.
 type sessionLifecycle struct {
 	id      string
+	newRoot bool // NewSession allocates a root ID before it may be persisted.
 	closing bool
 	active  int
 	pending sync.WaitGroup
@@ -78,6 +79,9 @@ func (a *Agent) beginSessionConstruction(ctx context.Context, sid string) (conte
 	if a.stopped {
 		return nil, nil, errors.New("agent stopped")
 	}
+	if a.deletion != nil {
+		return nil, nil, errors.New("session deletion in progress; retry after it finishes")
+	}
 	if a.team == nil {
 		return nil, nil, errors.New("agent not initialized")
 	}

pkg/session/delete_subtree_test.go

@@ -0,0 +1,133 @@
+package session
+
+import (
+	"context"
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDeleteSessionRemovesSubtreeAndStoredMedia(t *testing.T) {
+	t.Parallel()
+	manifestStores(t, func(t *testing.T, store Store, manifest GeneratedMediaManifest) {
+		t.Helper()
+		ctx := t.Context()
+		if sqlite, ok := store.(*SQLiteSessionStore); ok {
+			_, err := sqlite.db.ExecContext(ctx, "PRAGMA foreign_keys = ON")
+			require.NoError(t, err)
+		}
+		for _, entry := range []struct{ id, parent string }{{"root", ""}, {"child", "root"}, {"grandchild", "child"}, {"kept", ""}} {
+			require.NoError(t, store.AddSession(ctx, New(WithID(entry.id), WithParentID(entry.parent), WithUserMessage("history"))))
+			require.NoError(t, manifest.AddGeneratedFile(ctx, GeneratedFile{SessionID: entry.id, RelPath: "image.png", MimeType: "image/png"}))
+			require.NoError(t, manifest.(GeneratedMediaBlobStore).AddGeneratedBlob(ctx, entry.id, "image.png", []byte("bytes")))
+		}
+		require.NoError(t, store.DeleteSession(ctx, "root"))
+		for _, id := range []string{"root", "child", "grandchild"} {
+			_, err := store.GetSession(ctx, id)
+			require.ErrorIs(t, err, ErrNotFound)
+			_, err = manifest.LookupGeneratedFile(ctx, id, "image.png")
+			require.ErrorIs(t, err, ErrGeneratedFileNotFound)
+			_, err = manifest.(GeneratedMediaBlobStore).LookupGeneratedBlob(ctx, id, "image.png")
+			require.ErrorIs(t, err, ErrGeneratedBlobNotFound)
+		}
+		_, err := store.GetSession(ctx, "kept")
+		require.NoError(t, err)
+		_, err = manifest.LookupGeneratedFile(ctx, "kept", "image.png")
+		require.NoError(t, err)
+		_, err = manifest.(GeneratedMediaBlobStore).LookupGeneratedBlob(ctx, "kept", "image.png")
+		require.NoError(t, err)
+		require.ErrorIs(t, store.DeleteSession(ctx, "root"), ErrNotFound)
+	})
+}
+
+func TestDeleteSessionCanceledBeforeMutation(t *testing.T) {
+	t.Parallel()
+	manifestStores(t, func(t *testing.T, store Store, _ GeneratedMediaManifest) {
+		t.Helper()
+		require.NoError(t, store.AddSession(t.Context(), New(WithID("root"))))
+		require.NoError(t, store.AddSession(t.Context(), New(WithID("child"), WithParentID("root"))))
+		ctx, cancel := context.WithCancel(t.Context())
+		cancel()
+		require.ErrorIs(t, store.DeleteSession(ctx, "root"), context.Canceled)
+		for _, id := range []string{"root", "child"} {
+			_, err := store.GetSession(t.Context(), id)
+			require.NoError(t, err)
+		}
+	})
+}
+
+func TestSQLiteDeleteSubtreeRollback(t *testing.T) {
+	t.Parallel()
+	store := openMemoryStore(t)
+	ctx := t.Context()
+	require.NoError(t, store.AddSession(ctx, New(WithID("root"))))
+	require.NoError(t, store.AddSession(ctx, New(WithID("child"), WithParentID("root"))))
+	require.NoError(t, store.AddGeneratedBlob(ctx, "child", "image.png", []byte("kept")))
+	require.NoError(t, store.AddGeneratedFile(ctx, GeneratedFile{SessionID: "child", RelPath: "image.png"}))
+	_, err := store.db.ExecContext(ctx, `CREATE TRIGGER block_delete BEFORE DELETE ON sessions BEGIN SELECT RAISE(ABORT, 'delete blocked'); END`)
+	require.NoError(t, err)
+	require.ErrorContains(t, store.DeleteSession(ctx, "root"), "delete blocked")
+	_, err = store.GetSession(ctx, "root")
+	require.NoError(t, err)
+	_, err = store.GetSession(ctx, "child")
+	require.NoError(t, err)
+	blob, err := store.LookupGeneratedBlob(ctx, "child", "image.png")
+	require.NoError(t, err)
+	assert.Equal(t, []byte("kept"), blob)
+	_, err = store.LookupGeneratedFile(ctx, "child", "image.png")
+	require.NoError(t, err)
+}
+
+func TestDeleteSessionPrunesEmbeddedGrandchildMedia(t *testing.T) {
+	t.Parallel()
+	manifestStores(t, func(t *testing.T, store Store, manifest GeneratedMediaManifest) {
+		t.Helper()
+		root := New(WithID("root"))
+		require.NoError(t, store.AddSession(t.Context(), root))
+		child := New(WithID("child"), WithParentID(root.ID))
+		grandchild := New(WithID("grandchild"), WithParentID(child.ID))
+		child.AddSubSession(grandchild)
+		require.NoError(t, store.AddSubSession(t.Context(), root.ID, child))
+		require.NoError(t, manifest.AddGeneratedFile(t.Context(), GeneratedFile{SessionID: grandchild.ID, RelPath: "image.png"}))
+		blobs := manifest.(GeneratedMediaBlobStore)
+		require.NoError(t, blobs.AddGeneratedBlob(t.Context(), grandchild.ID, "image.png", []byte("bytes")))
+		require.NoError(t, store.DeleteSession(t.Context(), root.ID))
+		_, err := manifest.LookupGeneratedFile(t.Context(), grandchild.ID, "image.png")
+		require.ErrorIs(t, err, ErrGeneratedFileNotFound)
+		_, err = blobs.LookupGeneratedBlob(t.Context(), grandchild.ID, "image.png")
+		require.ErrorIs(t, err, ErrGeneratedBlobNotFound)
+	})
+}
+
+func TestDeleteSessionConcurrentUnrelatedChildRegistration(t *testing.T) {
+	t.Parallel()
+	store := NewInMemorySessionStore()
+	root := New(WithID("unrelated"))
+	require.NoError(t, store.AddSession(t.Context(), root))
+	var children []*Session
+	for i := range 100 {
+		child := New(WithID("child-" + strconv.Itoa(i)))
+		root.AddLiveSubSession(child)
+		children = append(children, child)
+		require.NoError(t, store.AddSession(t.Context(), New(WithID("delete-"+strconv.Itoa(i)))))
+	}
+	var wg sync.WaitGroup
+	wg.Go(func() {
+		for _, child := range children {
+			assert.NoError(t, store.AddSubSession(t.Context(), root.ID, child))
+		}
+	})
+	wg.Go(func() {
+		for i := range 100 {
+			assert.NoError(t, store.DeleteSession(t.Context(), "delete-"+strconv.Itoa(i)))
+		}
+	})
+	wg.Wait()
+	for _, child := range children {
+		_, err := store.GetSession(t.Context(), child.ID)
+		require.NoError(t, err)
+	}
+}

pkg/session/store.go

@@ -99,6 +99,8 @@ type Store interface {
 	GetSessionByOrigin(ctx context.Context, id, origin string) (*Session, error)
 	GetSessions(ctx context.Context) ([]*Session, error)
 	GetSessionSummaries(ctx context.Context) ([]Summary, error)
+	// DeleteSession removes the session subtree and its stored media, not workspace files.
+	// Callers must stop writers before deleting their sessions.
 	DeleteSession(ctx context.Context, id string) error
 	UpdateSession(ctx context.Context, session *Session) error // Updates metadata only (not messages/items)
 	SetSessionStarred(ctx context.Context, id string, starred bool) error
@@ -232,17 +234,54 @@ func (s *InMemorySessionStore) GetSessionSummaries(_ context.Context) ([]Summary
 	return summaries, nil
 }
 
-func (s *InMemorySessionStore) DeleteSession(_ context.Context, id string) error {
+func (s *InMemorySessionStore) DeleteSession(ctx context.Context, id string) error {
 	if id == "" {
 		return ErrEmptyID
 	}
-	_, exists := s.sessions.Load(id)
-	if !exists {
+	if _, exists := s.sessions.Load(id); !exists {
 		return ErrNotFound
 	}
-	s.sessions.Delete(id)
-	s.deleteGeneratedFiles(id)
-	s.deleteGeneratedBlobs(id)
+	children := make(map[string][]string)
+	var sessions []*Session
+	s.sessions.Range(func(_ string, sess *Session) bool {
+		sessions = append(sessions, sess)
+		return true
+	})
+	visited := make(map[*Session]bool)
+	for len(sessions) > 0 {
+		sess := sessions[len(sessions)-1]
+		sessions = sessions[:len(sessions)-1]
+		if visited[sess] {
+			continue
+		}
+		visited[sess] = true
+		sess.mu.RLock()
+		children[sess.ParentID] = append(children[sess.ParentID], sess.ID)
+		for _, item := range sess.Messages {
+			if item.SubSession != nil {
+				children[sess.ID] = append(children[sess.ID], item.SubSession.ID)
+				sessions = append(sessions, item.SubSession)
+			}
+		}
+		sess.mu.RUnlock()
+	}
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	pending := []string{id}
+	seen := make(map[string]struct{})
+	for len(pending) > 0 {
+		id := pending[len(pending)-1]
+		pending = pending[:len(pending)-1]
+		if _, visited := seen[id]; visited {
+			continue
+		}
+		seen[id] = struct{}{}
+		pending = append(pending, children[id]...)
+		s.sessions.Delete(id)
+		s.deleteGeneratedFiles(id)
+		s.deleteGeneratedBlobs(id)
+	}
 	return nil
 }
 
@@ -371,7 +410,9 @@ func (s *InMemorySessionStore) AddSubSession(_ context.Context, parentSessionID
 	if !exists {
 		return ErrNotFound
 	}
+	subSession.mu.Lock()
 	subSession.ParentID = parentSessionID
+	subSession.mu.Unlock()
 	subSession.syncEvaluationCosts()
 	s.sessions.Store(subSession.ID, subSession)
 	parent.AddSubSession(subSession)
@@ -1028,16 +1069,22 @@ func (s *SQLiteSessionStore) DeleteSession(ctx context.Context, id string) error
 	}
 	defer func() { _ = tx.Rollback() }()
 
+	// Prune descendants' non-FK media before cascading away their ancestry.
+	const subtree = `WITH RECURSIVE subtree(id) AS (
+		SELECT id FROM sessions WHERE id = ?
+		UNION
+		SELECT sessions.id FROM sessions JOIN subtree ON sessions.parent_id = subtree.id
+	) `
+	if _, err := tx.ExecContext(ctx, subtree+"DELETE FROM generated_media_blobs WHERE session_id IN (SELECT id FROM subtree)", id); err != nil {
+		return err
+	}
+	if _, err := tx.ExecContext(ctx, subtree+"DELETE FROM generated_media_manifest WHERE session_id IN (SELECT id FROM subtree)", id); err != nil {
+		return err
+	}
 	result, err := tx.ExecContext(ctx, "DELETE FROM sessions WHERE id = ?", id)
 	if err != nil {
 		return err
 	}
-	if _, err := tx.ExecContext(ctx, "DELETE FROM generated_media_blobs WHERE session_id = ?", id); err != nil {
-		return err
-	}
-	if _, err := tx.ExecContext(ctx, "DELETE FROM generated_media_manifest WHERE session_id = ?", id); err != nil {
-		return err
-	}
 
 	rowsAffected, err := result.RowsAffected()
 	if err != nil {