feat(sessions): per-session cleanup API + cleanup facts on read model (#2811)

backend/internal/cli/dto_drift_e2e_test.go

@@ -92,6 +92,10 @@ func (f *fakeSessionService) Cleanup(context.Context, domain.ProjectID) (session
 	return sessionsvc.CleanupOutcome{}, nil
 }
 
+func (f *fakeSessionService) CleanupSession(context.Context, domain.SessionID) (domain.SessionCleanupRecord, error) {
+	return domain.SessionCleanupRecord{}, nil
+}
+
 func (f *fakeSessionService) Rename(context.Context, domain.SessionID, string) error {
 	return nil
 }

backend/internal/daemon/daemon.go

@@ -22,6 +22,7 @@ import (
 	"github.com/aoagents/agent-orchestrator/backend/internal/httpd/controllers"
 	"github.com/aoagents/agent-orchestrator/backend/internal/mobilebridge"
 	"github.com/aoagents/agent-orchestrator/backend/internal/notify"
+	"github.com/aoagents/agent-orchestrator/backend/internal/observe/reconciler"
 	"github.com/aoagents/agent-orchestrator/backend/internal/ports"
 	"github.com/aoagents/agent-orchestrator/backend/internal/preview"
 	"github.com/aoagents/agent-orchestrator/backend/internal/push"
@@ -265,9 +266,10 @@ func Run() error {
 	}
 
 	// Reconcile sessions on boot: adopt crash-surviving runtimes, capture and
-	// terminate dead ones, reap leaked tmux, then restore shutdown-saved
-	// sessions. Best-effort: a failure is logged but never blocks boot. Placed
-	// before srv.Run so sessions are consistent before the server serves.
+	// terminate dead ones, then restore shutdown-saved sessions. Run to
+	// completion (synchronous, as before) so restore-pending sessions are settled
+	// before the terminal-resource reconciler starts and before the server
+	// serves. Best-effort: a failure is logged but never blocks boot.
 	if reconcileErr := sessMgr.Reconcile(ctx); reconcileErr != nil {
 		log.Error("reconcile sessions on boot failed", "err", reconcileErr)
 	}
@@ -281,6 +283,16 @@ func Run() error {
 	// the recovery sweep is the backstop if it does not finish before shutdown.
 	go lcStack.LCM.DispatchAllPendingWorkerIdleEvents(ctx)
 
+	// Start the terminal-resource reconciler AFTER Reconcile/RestoreAll has
+	// settled: it subscribes to the CDC session_updated wake, then drains the
+	// sessions-driven candidate backlog (the pre-existing leaked worktrees #2811
+	// is about) onto its async worker, and retries capped-backoff pending
+	// cleanups on a sweep. Subscribing before its own boot snapshot closes the
+	// boot-window race; draining is async so a large backlog never delays serving.
+	// Its done-channel joins lcStack.Stop so it unsubscribes before cdcPipe.Stop.
+	termReconciler := reconciler.New(sessMgr, store, cdcPipe.Broadcaster, reconciler.Config{Logger: log})
+	lcStack.reconcilerDone = termReconciler.Start(ctx)
+
 	// ponytail: 5s tolerates a brief frontend restart; tune if dev hot-reload trips it.
 	const supervisorGrace = 5 * time.Second
 

backend/internal/daemon/lifecycle_wiring.go

@@ -43,6 +43,11 @@ type lifecycleStack struct {
 	scmDone       <-chan struct{}
 	trackerDone   <-chan struct{}
 	sweepDone     <-chan struct{}
+	// reconcilerDone is the terminal-resource reconciler's drain channel. Assigned
+	// after boot Reconcile completes (daemon.go); nil until then, so Stop must
+	// nil-guard it. It is drained BEFORE the CDC pipe stops so the reconciler
+	// unsubscribes before a late Publish could send on its closed queue.
+	reconcilerDone <-chan struct{}
 }
 
 // workerIdleSweepInterval is the low-frequency recovery cadence that redelivers
@@ -127,6 +132,11 @@ func (l *lifecycleStack) Stop() {
 	if l.trackerDone != nil {
 		<-l.trackerDone
 	}
+	// Drained before the CDC pipe stops (daemon shutdown order) so the reconciler
+	// unsubscribes before a late Publish could send on its closed queue.
+	if l.reconcilerDone != nil {
+		<-l.reconcilerDone
+	}
 }
 
 // sessionLifecycle is the narrow surface of sessionmanager.Manager used for
@@ -141,6 +151,10 @@ type sessionLifecycle interface {
 	Reconcile(ctx context.Context) error
 	RestoreAll(ctx context.Context) error
 	Kill(ctx context.Context, id domain.SessionID) (bool, error)
+	// FinalizeTerminalSession is the terminal-resource reconciler's per-session
+	// unit of work (release a terminated session's runtime + workspace). Exposed
+	// here so the daemon can hand the manager to the reconciler runner.
+	FinalizeTerminalSession(ctx context.Context, id domain.SessionID) error
 }
 
 // startSession builds the controller-facing session service: a session manager
@@ -172,13 +186,29 @@ func startSession(cfg config.Config, runtime runtimeselect.Runtime, store *sqlit
 		Scratch:  scratchWS,
 		Projects: store,
 	})
+	// Build the review engine before the session manager so the manager can gate
+	// worktree reclaim on a worker's reviewer pane (destroy it + confirm it's gone
+	// before removing the worktree). The engine depends only on the store/launcher,
+	// not on the manager, so there is no construction cycle.
+	reviewers, err := reviewer.NewResolver()
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("reviewer resolver: %w", err)
+	}
+	reviewEngine := reviewcore.New(reviewcore.Deps{
+		Store:    store,
+		Sessions: store,
+		PRs:      store,
+		Projects: store,
+		Launcher: reviewcore.NewLauncher(reviewers, runtime, cfg.DataDir),
+	})
 	mgr := sessionmanager.New(sessionmanager.Deps{
 		Runtime:   runtime,
 		Agents:    agents,
 		Workspace: ws,
 		Store:     store,
 		Messenger: messenger,
 		Lifecycle: lcm,
+		Reviewer:  reviewEngine,
 		DataDir:   cfg.DataDir,
 		Logger:    log,
 	})
@@ -213,18 +243,8 @@ func startSession(cfg config.Config, runtime runtimeselect.Runtime, store *sqlit
 	// Triggering a review spawns a reviewer over the worker's worktree, resolved
 	// from the reviewer registry (distinct from the worker agent set). The
 	// reviewer posts its review to the PR itself, so the service needs no SCM
-	// writer.
-	reviewers, err := reviewer.NewResolver()
-	if err != nil {
-		return nil, nil, nil, fmt.Errorf("reviewer resolver: %w", err)
-	}
-	reviewEngine := reviewcore.New(reviewcore.Deps{
-		Store:    store,
-		Sessions: store,
-		PRs:      store,
-		Projects: store,
-		Launcher: reviewcore.NewLauncher(reviewers, runtime, cfg.DataDir),
-	})
+	// writer. The engine itself was built above (the manager reuses it for the
+	// reviewer-pane teardown gate).
 	reviewSvc := reviewsvc.New(reviewEngine, store, reviewsvc.WithLifecycleReducer(lcm))
 	return sessionSvc, reviewSvc, mgr, nil
 }

backend/internal/daemon/wiring_test.go

@@ -633,6 +633,10 @@ func (f *fakeSessionLifecycle) RestoreAll(_ context.Context) error {
 	return f.restoreErr
 }
 
+func (f *fakeSessionLifecycle) FinalizeTerminalSession(_ context.Context, _ domain.SessionID) error {
+	return nil
+}
+
 // TestWiring_SessionLifecycleInterfaceInvokedByDaemon asserts the
 // sessionLifecycle interface is satisfied by *sessionmanager.Manager (compile
 // check) and that Reconcile and RestoreAll dispatch correctly through the

backend/internal/domain/session.go

@@ -85,4 +85,8 @@ type Session struct {
 	// They feed status derivation and are surfaced on the API read model. Not
 	// serialized here: the HTTP boundary maps them to the curated wire shape.
 	PRs []PRFacts `json:"-"`
+	// Cleanup holds the terminal-resource cleanup facts for a terminated session,
+	// nil when none exist yet (a live session, or one not yet finalized). Joined
+	// at read time; the HTTP boundary maps it to the curated wire shape.
+	Cleanup *SessionCleanupRecord `json:"-"`
 }

backend/internal/httpd/apispec/openapi.yaml

@@ -1225,6 +1225,45 @@ paths:
       summary: Report an agent activity-state signal for a session
       tags:
       - sessions
+  /api/v1/sessions/{sessionId}/cleanup:
+    post:
+      operationId: cleanupSession
+      parameters:
+      - description: Session identifier, e.g. project-1.
+        in: path
+        name: sessionId
+        required: true
+        schema:
+          description: Session identifier, e.g. project-1.
+          type: string
+      responses:
+        "200":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/CleanupSessionResponse'
+          description: OK
+        "404":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/APIError'
+          description: Not Found
+        "409":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/APIError'
+          description: Conflict
+        "500":
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/APIError'
+          description: Internal Server Error
+      summary: Reclaim a single terminated session's runtime + workspace resources
+      tags:
+      - sessions
   /api/v1/sessions/{sessionId}/kill:
     post:
       operationId: killSession
@@ -2319,6 +2358,21 @@ components:
       - branchChanged
       - takenOverFrom
       type: object
+    CleanupSessionResponse:
+      properties:
+        cleanup:
+          $ref: '#/components/schemas/SessionCleanupView'
+        isTerminated:
+          type: boolean
+        ok:
+          type: boolean
+        sessionId:
+          type: string
+      required:
+      - ok
+      - sessionId
+      - isTerminated
+      type: object
     CleanupSessionsResponse:
       properties:
         cleaned:
@@ -2352,6 +2406,8 @@ components:
           $ref: '#/components/schemas/DomainActivity'
         branch:
           type: string
+        cleanup:
+          $ref: '#/components/schemas/SessionCleanupView'
         createdAt:
           format: date-time
           type: string
@@ -3240,6 +3296,34 @@ components:
       - sessionId
       - message
       type: object
+    SessionCleanupView:
+      properties:
+        attemptCount:
+          format: int64
+          type: integer
+        failureCode:
+          type: string
+        nextAttemptAt:
+          format: date-time
+          type:
+          - "null"
+          - string
+        runtimeReleasedAt:
+          format: date-time
+          type:
+          - "null"
+          - string
+        workspaceDisposition:
+          enum:
+          - pending
+          - removed
+          - preserved_dirty
+          - failed
+          - not_applicable
+          type: string
+      required:
+      - workspaceDisposition
+      type: object
     SessionPRCISummary:
       properties:
         failingChecks:

backend/internal/httpd/apispec/specgen/build.go

@@ -167,6 +167,8 @@ var schemaNames = map[string]string{
 	"ControllersWorkspaceFileSummary":             "WorkspaceFileSummary",
 	"ControllersWorkspaceFileResponse":            "WorkspaceFileResponse",
 	"ControllersKillSessionResponse":              "KillSessionResponse",
+	"ControllersCleanupSessionResponse":           "CleanupSessionResponse",
+	"ControllersSessionCleanupView":               "SessionCleanupView",
 	"ControllersRollbackSessionResponse":          "RollbackSessionResponse",
 	"ControllersSendSessionMessageRequest":        "SendSessionMessageRequest",
 	"ControllersSendSessionMessageResponse":       "SendSessionMessageResponse",
@@ -947,6 +949,17 @@ func sessionOperations() []operation {
 				{http.StatusInternalServerError, envelope.APIError{}},
 			},
 		},
+		{
+			method: http.MethodPost, path: "/api/v1/sessions/{sessionId}/cleanup", id: "cleanupSession", tag: "sessions",
+			summary:    "Reclaim a single terminated session's runtime + workspace resources",
+			pathParams: []any{controllers.SessionIDParam{}},
+			resps: []respUnit{
+				{http.StatusOK, controllers.CleanupSessionResponse{}},
+				{http.StatusNotFound, envelope.APIError{}},
+				{http.StatusConflict, envelope.APIError{}},
+				{http.StatusInternalServerError, envelope.APIError{}},
+			},
+		},
 		{
 			method: http.MethodPost, path: "/api/v1/sessions/{sessionId}/rollback", id: "rollbackSession", tag: "sessions",
 			summary:    "Undo a partially-completed spawn (delete seed row, or kill if spawn output exists)",

backend/internal/httpd/controllers/dto.go

@@ -143,6 +143,31 @@ type SessionView struct {
 	// Metadata.
 	PreviewRevision int64            `json:"previewRevision,omitempty"`
 	PRs             []SessionPRFacts `json:"prs"`
+	// Cleanup is the terminal-resource cleanup state for a terminated session
+	// (runtime + workspace release progress). Omitted for a live session or one
+	// with no facts row yet. Derived at read time — no display status is stored.
+	Cleanup *SessionCleanupView `json:"cleanup,omitempty"`
+}
+
+// SessionCleanupView is the read-model view of a terminated session's
+// terminal-resource cleanup facts. Every field is a durable fact, never a
+// derived display status: the client renders cleanup state (cleaning up /
+// worktree kept — uncommitted changes / cleanup failed / done) from
+// workspaceDisposition.
+type SessionCleanupView struct {
+	// WorkspaceDisposition is the rollup: pending, removed, preserved_dirty,
+	// failed, or not_applicable.
+	WorkspaceDisposition domain.WorkspaceDisposition `json:"workspaceDisposition" enum:"pending,removed,preserved_dirty,failed,not_applicable"`
+	// RuntimeReleasedAt is when the runtime handle was genuinely released; omitted
+	// when the runtime has not been confirmed released.
+	RuntimeReleasedAt *time.Time `json:"runtimeReleasedAt,omitempty"`
+	// AttemptCount is how many teardown attempts have run so far.
+	AttemptCount int64 `json:"attemptCount,omitempty"`
+	// NextAttemptAt is when the next capped-backoff retry is due; omitted when
+	// none is scheduled (a terminal disposition, or preserved-dirty pause).
+	NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"`
+	// FailureCode is a machine code for the last teardown failure; omitted when none.
+	FailureCode string `json:"failureCode,omitempty"`
 }
 
 // ListSessionsResponse is the body of GET /api/v1/sessions.
@@ -287,6 +312,17 @@ type KillSessionResponse struct {
 	Freed     bool             `json:"freed,omitempty"`
 }
 
+// CleanupSessionResponse is the body of POST /api/v1/sessions/{sessionId}/cleanup:
+// the resulting cleanup facts after reclaiming one terminated session's runtime +
+// workspace. isTerminated is always true here (the endpoint 409s for a live
+// session); it is surfaced for parity with the session read model.
+type CleanupSessionResponse struct {
+	OK           bool                `json:"ok"`
+	SessionID    domain.SessionID    `json:"sessionId"`
+	IsTerminated bool                `json:"isTerminated"`
+	Cleanup      *SessionCleanupView `json:"cleanup,omitempty"`
+}
+
 // RollbackSessionResponse is the body of POST /api/v1/sessions/{sessionId}/rollback.
 // Exactly one of Deleted/Killed is true on a successful rollback; both are
 // false when the session was already absent or already terminated (benign).