From 424e6e824bd743cfad038ca05085e41c34c4f992 Mon Sep 17 00:00:00 2001 From: prateek Date: Mon, 1 Jun 2026 23:31:21 +0530 Subject: [PATCH] refactor: move session status assembly to service (#62) (#67) Co-authored-by: itrytoohard --- README.md | 2 +- backend/internal/domain/status.go | 49 --- backend/internal/domain/status_test.go | 38 --- backend/internal/httpd/api.go | 6 + backend/internal/httpd/apispec/openapi.yaml | 323 ++++++++++++++++++ .../internal/httpd/controllers/sessions.go | 276 +++++++++++++++ .../httpd/controllers/sessions_test.go | 174 ++++++++++ .../integration/lifecycle_sqlite_test.go | 8 +- backend/internal/service/session.go | 143 ++++++++ backend/internal/service/session_status.go | 49 +++ .../internal/service/session_status_test.go | 42 +++ backend/internal/service/session_test.go | 68 ++++ .../{session => session_manager}/manager.go | 86 ++--- .../manager_test.go | 22 +- docs/README.md | 2 +- docs/architecture.md | 19 +- docs/status.md | 10 +- 17 files changed, 1138 insertions(+), 179 deletions(-) delete mode 100644 backend/internal/domain/status_test.go create mode 100644 backend/internal/httpd/controllers/sessions.go create mode 100644 backend/internal/httpd/controllers/sessions_test.go create mode 100644 backend/internal/service/session.go create mode 100644 backend/internal/service/session_status.go create mode 100644 backend/internal/service/session_status_test.go create mode 100644 backend/internal/service/session_test.go rename backend/internal/{session => session_manager}/manager.go (76%) rename backend/internal/{session => session_manager}/manager_test.go (92%) diff --git a/README.md b/README.md index 33e08dae8..146087bde 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Rewrite of the agent-orchestrator: a long-running Go backend daemon (`backend/`) paired with a placeholder Electron + TypeScript frontend shell (`frontend/`). See [`docs/`](docs/README.md) for architecture and status — start with the -Lifecycle Manager + Session Manager lane in [`docs/architecture.md`](docs/architecture.md). +Lifecycle Manager + Session Service lane in [`docs/architecture.md`](docs/architecture.md). ## Backend daemon diff --git a/backend/internal/domain/status.go b/backend/internal/domain/status.go index ad8846855..e42a0ed0d 100644 --- a/backend/internal/domain/status.go +++ b/backend/internal/domain/status.go @@ -19,52 +19,3 @@ const ( StatusIdle SessionStatus = "idle" StatusTerminated SessionStatus = "terminated" ) - -// DeriveStatus is the ONLY producer of display status. It is a pure function of -// persisted session facts and PR facts: is_terminated, activity_state, and the PR -// table are the durable facts that tell the UI what it needs to know. -func DeriveStatus(rec SessionRecord, pr *PRFacts) SessionStatus { - if rec.IsTerminated { - if pr != nil && pr.Merged { - return StatusMerged - } - return StatusTerminated - } - - if rec.Activity.State == ActivityWaitingInput { - return StatusNeedsInput - } - - if pr != nil { - if pr.Merged { - return StatusMerged - } - if !pr.Closed { - return prPipelineStatus(*pr) - } - } - - if rec.Activity.State == ActivityActive { - return StatusWorking - } - return StatusIdle -} - -func prPipelineStatus(pr PRFacts) SessionStatus { - switch { - case pr.CI == CIFailing: - return StatusCIFailed - case pr.Draft: - return StatusDraft - case pr.Review == ReviewChangesRequest || pr.ReviewComments: - return StatusChangesRequested - case pr.Mergeability == MergeMergeable: - return StatusMergeable - case pr.Review == ReviewApproved: - return StatusApproved - case pr.Review == ReviewRequired: - return StatusReviewPending - default: - return StatusPROpen - } -} diff --git a/backend/internal/domain/status_test.go b/backend/internal/domain/status_test.go deleted file mode 100644 index 075098687..000000000 --- a/backend/internal/domain/status_test.go +++ /dev/null @@ -1,38 +0,0 @@ -package domain - -import "testing" - -func rec(activity ActivityState, terminated bool) SessionRecord { - return SessionRecord{Activity: Activity{State: activity}, IsTerminated: terminated} -} - -func pr(facts PRFacts) *PRFacts { return &facts } - -func TestDeriveStatusFromSessionFactsAndPR(t *testing.T) { - tests := []struct { - name string - rec SessionRecord - pr *PRFacts - want SessionStatus - }{ - {"terminated", rec(ActivityExited, true), nil, StatusTerminated}, - {"merged-pr", rec(ActivityIdle, true), pr(PRFacts{Merged: true}), StatusMerged}, - {"needs-input", rec(ActivityWaitingInput, false), pr(PRFacts{CI: CIFailing}), StatusNeedsInput}, - {"ci-failed", rec(ActivityIdle, false), pr(PRFacts{CI: CIFailing}), StatusCIFailed}, - {"draft", rec(ActivityIdle, false), pr(PRFacts{Draft: true}), StatusDraft}, - {"changes-requested", rec(ActivityIdle, false), pr(PRFacts{Review: ReviewChangesRequest}), StatusChangesRequested}, - {"mergeable", rec(ActivityIdle, false), pr(PRFacts{Mergeability: MergeMergeable}), StatusMergeable}, - {"approved", rec(ActivityIdle, false), pr(PRFacts{Review: ReviewApproved}), StatusApproved}, - {"review-pending", rec(ActivityIdle, false), pr(PRFacts{Review: ReviewRequired}), StatusReviewPending}, - {"pr-open", rec(ActivityIdle, false), pr(PRFacts{}), StatusPROpen}, - {"working", rec(ActivityActive, false), nil, StatusWorking}, - {"idle", rec(ActivityIdle, false), nil, StatusIdle}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := DeriveStatus(tt.rec, tt.pr); got != tt.want { - t.Fatalf("got %q, want %q", got, tt.want) - } - }) - } -} diff --git a/backend/internal/httpd/api.go b/backend/internal/httpd/api.go index 9480cdad4..a61ba7849 100644 --- a/backend/internal/httpd/api.go +++ b/backend/internal/httpd/api.go @@ -19,6 +19,7 @@ import ( // registered but returns the OpenAPI-backed 501 response. type APIDeps struct { Projects project.Manager + Sessions controllers.SessionService } // API owns one controller per resource and is the single Register call the @@ -26,6 +27,7 @@ type APIDeps struct { type API struct { cfg config.Config projects *controllers.ProjectsController + sessions *controllers.SessionsController } // NewAPI constructs the API surface from its dependencies. cfg carries the @@ -37,6 +39,9 @@ func NewAPI(cfg config.Config, deps APIDeps) *API { projects: &controllers.ProjectsController{ Mgr: deps.Projects, }, + sessions: &controllers.SessionsController{ + Svc: deps.Sessions, + }, } } @@ -55,6 +60,7 @@ func (a *API) Register(root chi.Router) { r.Group(func(r chi.Router) { r.Use(middleware.Timeout(timeout)) a.projects.Register(r) + a.sessions.Register(r) // Sibling REST controllers plug in here. }) // Surfaces that intentionally bypass the REST timeout register at this level. diff --git a/backend/internal/httpd/apispec/openapi.yaml b/backend/internal/httpd/apispec/openapi.yaml index 970ec7f48..b626c5788 100644 --- a/backend/internal/httpd/apispec/openapi.yaml +++ b/backend/internal/httpd/apispec/openapi.yaml @@ -15,6 +15,8 @@ servers: tags: - name: projects description: Project registry, configuration, and lifecycle administration + - name: sessions + description: Agent session lifecycle and messaging paths: /api/v1/projects: @@ -228,6 +230,219 @@ paths: "404": { $ref: "#/components/responses/ProjectNotFound" } "501": { $ref: "#/components/responses/NotImplemented" } + /api/v1/sessions: + get: + operationId: listSessions + tags: [sessions] + summary: List sessions + parameters: + - name: project + in: query + schema: { type: string } + - name: active + in: query + schema: { type: boolean } + - name: orchestratorOnly + in: query + schema: { type: boolean } + - name: fresh + in: query + schema: { type: boolean } + responses: + "200": + description: Sessions listed + content: + application/json: + schema: + type: object + required: [sessions] + properties: + sessions: + type: array + items: { $ref: "#/components/schemas/Session" } + "400": + description: Bad request + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + "501": { $ref: "#/components/responses/NotImplemented" } + post: + operationId: spawnSession + tags: [sessions] + summary: Spawn a new agent session + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/SpawnSessionRequest" } + responses: + "201": + description: Session spawned + content: + application/json: + schema: + type: object + required: [session] + properties: + session: { $ref: "#/components/schemas/Session" } + "400": + description: Bad request + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + examples: + invalidJson: { value: { error: bad_request, code: INVALID_JSON, message: "Invalid JSON body" } } + projectRequired: { value: { error: bad_request, code: PROJECT_ID_REQUIRED, message: "projectId is required" } } + promptTooLong: { value: { error: bad_request, code: PROMPT_TOO_LONG, message: "prompt is too long" } } + "404": { $ref: "#/components/responses/ProjectNotFound" } + "500": { $ref: "#/components/responses/SessionOperationFailed" } + "501": { $ref: "#/components/responses/NotImplemented" } + + /api/v1/sessions/{sessionId}: + parameters: + - $ref: "#/components/parameters/SessionIDPath" + get: + operationId: getSession + tags: [sessions] + summary: Fetch one session + responses: + "200": + description: Session fetched + content: + application/json: + schema: + type: object + required: [session] + properties: + session: { $ref: "#/components/schemas/Session" } + "404": { $ref: "#/components/responses/SessionNotFound" } + "501": { $ref: "#/components/responses/NotImplemented" } + patch: + operationId: renameSession + tags: [sessions] + summary: Rename a session display name + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [displayName] + properties: + displayName: { type: string, minLength: 1 } + responses: + "200": + description: Session renamed + content: + application/json: + schema: + type: object + required: [session] + properties: + session: { $ref: "#/components/schemas/Session" } + "400": + description: Bad request + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + "404": { $ref: "#/components/responses/SessionNotFound" } + "501": { $ref: "#/components/responses/NotImplemented" } + + /api/v1/sessions/{sessionId}/restore: + parameters: + - $ref: "#/components/parameters/SessionIDPath" + post: + operationId: restoreSession + tags: [sessions] + summary: Restore a terminated session + responses: + "200": + description: Session restored + content: + application/json: + schema: + type: object + required: [ok, sessionId, session] + properties: + ok: { type: boolean } + sessionId: { type: string } + session: { $ref: "#/components/schemas/Session" } + "404": { $ref: "#/components/responses/SessionNotFound" } + "409": { $ref: "#/components/responses/SessionConflict" } + "501": { $ref: "#/components/responses/NotImplemented" } + + /api/v1/sessions/{sessionId}/kill: + parameters: + - $ref: "#/components/parameters/SessionIDPath" + post: + operationId: killSession + tags: [sessions] + summary: Mark a session terminated and tear down runtime/workspace resources + responses: + "200": + description: Kill attempted + content: + application/json: + schema: { $ref: "#/components/schemas/KillSessionResponse" } + "404": { $ref: "#/components/responses/SessionNotFound" } + "409": { $ref: "#/components/responses/SessionConflict" } + "501": { $ref: "#/components/responses/NotImplemented" } + + /api/v1/sessions/{sessionId}/send: + parameters: + - $ref: "#/components/parameters/SessionIDPath" + post: + operationId: sendSessionMessage + tags: [sessions] + summary: Send a message to a running session's agent + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/SendSessionMessageRequest" } + responses: + "200": + description: Message accepted + content: + application/json: + schema: { $ref: "#/components/schemas/SendSessionMessageResponse" } + "400": + description: Bad request + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + examples: + invalidJson: { value: { error: bad_request, code: INVALID_JSON, message: "Invalid JSON body" } } + messageRequired: { value: { error: bad_request, code: MESSAGE_REQUIRED, message: "Message is required" } } + "404": { $ref: "#/components/responses/SessionNotFound" } + "500": { $ref: "#/components/responses/SessionOperationFailed" } + "501": { $ref: "#/components/responses/NotImplemented" } + + /api/v1/orchestrators: + post: + operationId: spawnOrchestrator + tags: [sessions] + summary: Spawn an orchestrator session + requestBody: + required: true + content: + application/json: + schema: { $ref: "#/components/schemas/SpawnOrchestratorRequest" } + responses: + "201": + description: Orchestrator spawned + content: + application/json: + schema: { $ref: "#/components/schemas/SpawnOrchestratorResponse" } + "400": + description: Bad request + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + "404": { $ref: "#/components/responses/ProjectNotFound" } + "500": { $ref: "#/components/responses/SessionOperationFailed" } + "501": { $ref: "#/components/responses/NotImplemented" } + components: parameters: ProjectIDPath: @@ -237,6 +452,13 @@ components: schema: { type: string, minLength: 1 } description: Project identifier (registry key). + SessionIDPath: + name: sessionId + in: path + required: true + schema: { type: string, minLength: 1 } + description: Session identifier, e.g. project-1. + responses: NotImplemented: description: | @@ -255,6 +477,29 @@ components: schema: { $ref: "#/components/schemas/APIError" } example: { error: not_found, code: PROJECT_NOT_FOUND, message: "Unknown project" } + SessionNotFound: + description: Session not found + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + example: { error: not_found, code: SESSION_NOT_FOUND, message: "Unknown session" } + + SessionConflict: + description: Session is not in a valid state for the requested operation + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + examples: + notRestorable: { value: { error: conflict, code: SESSION_NOT_RESTORABLE, message: "Session is not restorable" } } + incompleteHandle: { value: { error: conflict, code: SESSION_INCOMPLETE_HANDLE, message: "Session is missing runtime or workspace handles" } } + + SessionOperationFailed: + description: Session operation failed + content: + application/json: + schema: { $ref: "#/components/schemas/APIError" } + example: { error: internal, code: SESSION_OPERATION_FAILED, message: "Session operation failed" } + schemas: APIError: type: object @@ -369,6 +614,84 @@ components: projectCount: { type: integer } degradedCount: { type: integer } + + Session: + type: object + required: [id, projectId, kind, activity, isTerminated, createdAt, updatedAt, status] + properties: + id: { type: string } + projectId: { type: string } + issueId: { type: string } + kind: { type: string, enum: [worker, orchestrator] } + harness: { type: string, enum: ["", claude-code, codex, aider, opencode] } + activity: { $ref: "#/components/schemas/SessionActivity" } + isTerminated: { type: boolean } + createdAt: { type: string, format: date-time } + updatedAt: { type: string, format: date-time } + status: + type: string + enum: [working, pr_open, draft, ci_failed, review_pending, changes_requested, approved, mergeable, merged, needs_input, idle, terminated] + + SessionActivity: + type: object + required: [state, lastActivityAt] + properties: + state: { type: string, enum: [active, idle, waiting_input, exited] } + lastActivityAt: { type: string, format: date-time } + + SpawnSessionRequest: + type: object + required: [projectId] + properties: + projectId: { type: string } + issueId: { type: string } + kind: { type: string, enum: [worker, orchestrator], default: worker } + harness: { type: string, enum: ["", claude-code, codex, aider, opencode] } + branch: { type: string } + prompt: { type: string, maxLength: 4096 } + agentRules: { type: string } + + SendSessionMessageRequest: + type: object + required: [message] + properties: + message: { type: string, minLength: 1, maxLength: 4096 } + + SendSessionMessageResponse: + type: object + required: [ok, sessionId, message] + properties: + ok: { type: boolean } + sessionId: { type: string } + message: { type: string } + + KillSessionResponse: + type: object + required: [ok, sessionId] + properties: + ok: { type: boolean } + sessionId: { type: string } + freed: { type: boolean } + + SpawnOrchestratorRequest: + type: object + required: [projectId] + properties: + projectId: { type: string } + clean: { type: boolean, default: false } + + SpawnOrchestratorResponse: + type: object + required: [orchestrator] + properties: + orchestrator: + type: object + required: [id, projectId] + properties: + id: { type: string } + projectId: { type: string } + projectName: { type: string } + # ---- Behaviour config blobs ---- TrackerConfig: diff --git a/backend/internal/httpd/controllers/sessions.go b/backend/internal/httpd/controllers/sessions.go new file mode 100644 index 000000000..6beccf8e3 --- /dev/null +++ b/backend/internal/httpd/controllers/sessions.go @@ -0,0 +1,276 @@ +package controllers + +import ( + "context" + "errors" + "net/http" + "strconv" + "strings" + "unicode" + + "github.com/go-chi/chi/v5" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/httpd/apispec" + "github.com/aoagents/agent-orchestrator/backend/internal/httpd/envelope" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" + "github.com/aoagents/agent-orchestrator/backend/internal/service" + sessionmanager "github.com/aoagents/agent-orchestrator/backend/internal/session_manager" +) + +const ( + maxPromptLen = 4096 + maxMessageLen = 4096 +) + +// SessionService is the controller-facing session service contract. +type SessionService interface { + List(ctx context.Context, filter service.SessionListFilter) ([]domain.Session, error) + Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Session, error) + Get(ctx context.Context, id domain.SessionID) (domain.Session, error) + Restore(ctx context.Context, id domain.SessionID) (domain.Session, error) + Kill(ctx context.Context, id domain.SessionID) (bool, error) + Send(ctx context.Context, id domain.SessionID, message string) error +} + +// SessionsController owns the session routes. Nil keeps routes registered but +// returns OpenAPI-backed 501s. +type SessionsController struct { + Svc SessionService +} + +// Register mounts the session routes on the supplied router. +func (c *SessionsController) Register(r chi.Router) { + r.Get("/sessions", c.list) + r.Post("/sessions", c.spawn) + r.Get("/sessions/{sessionId}", c.get) + r.Patch("/sessions/{sessionId}", c.rename) + r.Post("/sessions/{sessionId}/restore", c.restore) + r.Post("/sessions/{sessionId}/kill", c.kill) + r.Post("/sessions/{sessionId}/send", c.send) + r.Post("/orchestrators", c.spawnOrchestrator) +} + +func (c *SessionsController) list(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "GET", "/api/v1/sessions") + return + } + filter, err := parseSessionListFilter(r) + if err != nil { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_QUERY", err.Error(), nil) + return + } + sessions, err := c.Svc.List(r.Context(), filter) + if err != nil { + writeSessionError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, map[string]any{"sessions": sessions}) +} + +type spawnSessionRequest struct { + ProjectID domain.ProjectID `json:"projectId"` + IssueID domain.IssueID `json:"issueId"` + Kind domain.SessionKind `json:"kind"` + Harness domain.AgentHarness `json:"harness"` + Branch string `json:"branch"` + Prompt string `json:"prompt"` + AgentRules string `json:"agentRules"` +} + +func (c *SessionsController) spawn(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/sessions") + return + } + var in spawnSessionRequest + if err := decodeJSON(r, &in); err != nil { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_JSON", "Invalid JSON body", nil) + return + } + if in.ProjectID == "" { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "PROJECT_ID_REQUIRED", "projectId is required", nil) + return + } + if len(in.Prompt) > maxPromptLen { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "PROMPT_TOO_LONG", "prompt is too long", nil) + return + } + if in.Kind == "" { + in.Kind = domain.KindWorker + } + sess, err := c.Svc.Spawn(r.Context(), ports.SpawnConfig{ProjectID: in.ProjectID, IssueID: in.IssueID, Kind: in.Kind, Harness: in.Harness, Branch: in.Branch, Prompt: in.Prompt, AgentRules: in.AgentRules}) + if err != nil { + writeSessionError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusCreated, map[string]any{"session": sess}) +} + +func (c *SessionsController) get(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "GET", "/api/v1/sessions/{sessionId}") + return + } + sess, err := c.Svc.Get(r.Context(), sessionID(r)) + if err != nil { + writeSessionError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, map[string]any{"session": sess}) +} + +func (c *SessionsController) rename(w http.ResponseWriter, r *http.Request) { + apispec.NotImplemented(w, r, "PATCH", "/api/v1/sessions/{sessionId}") +} + +func (c *SessionsController) restore(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/{sessionId}/restore") + return + } + sess, err := c.Svc.Restore(r.Context(), sessionID(r)) + if err != nil { + writeSessionError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, map[string]any{"ok": true, "sessionId": sessionID(r), "session": sess}) +} + +func (c *SessionsController) kill(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/{sessionId}/kill") + return + } + freed, err := c.Svc.Kill(r.Context(), sessionID(r)) + if err != nil { + writeSessionError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, map[string]any{"ok": true, "sessionId": sessionID(r), "freed": freed}) +} + +type sendSessionRequest struct { + Message string `json:"message"` +} + +func (c *SessionsController) send(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/sessions/{sessionId}/send") + return + } + var in sendSessionRequest + if err := decodeJSON(r, &in); err != nil { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_JSON", "Invalid JSON body", nil) + return + } + if in.Message == "" { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "MESSAGE_REQUIRED", "Message is required", nil) + return + } + if len(in.Message) > maxMessageLen { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "MESSAGE_TOO_LONG", "Message is too long", nil) + return + } + message := stripUnsafeControlChars(in.Message) + if err := c.Svc.Send(r.Context(), sessionID(r), message); err != nil { + writeSessionError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusOK, map[string]any{"ok": true, "sessionId": sessionID(r), "message": message}) +} + +type spawnOrchestratorRequest struct { + ProjectID domain.ProjectID `json:"projectId"` + Clean bool `json:"clean"` +} + +func (c *SessionsController) spawnOrchestrator(w http.ResponseWriter, r *http.Request) { + if c.Svc == nil { + apispec.NotImplemented(w, r, "POST", "/api/v1/orchestrators") + return + } + var in spawnOrchestratorRequest + if err := decodeJSON(r, &in); err != nil { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "INVALID_JSON", "Invalid JSON body", nil) + return + } + if in.ProjectID == "" { + envelope.WriteAPIError(w, r, http.StatusBadRequest, "bad_request", "PROJECT_ID_REQUIRED", "projectId is required", nil) + return + } + if in.Clean { + active := true + orchestrators, err := c.Svc.List(r.Context(), service.SessionListFilter{ProjectID: in.ProjectID, Active: &active, OrchestratorOnly: true}) + if err != nil { + writeSessionError(w, r, err) + return + } + for _, existing := range orchestrators { + if _, err := c.Svc.Kill(r.Context(), existing.ID); err != nil { + writeSessionError(w, r, err) + return + } + } + } + sess, err := c.Svc.Spawn(r.Context(), ports.SpawnConfig{ProjectID: in.ProjectID, Kind: domain.KindOrchestrator}) + if err != nil { + writeSessionError(w, r, err) + return + } + envelope.WriteJSON(w, http.StatusCreated, map[string]any{"orchestrator": map[string]any{"id": sess.ID, "projectId": sess.ProjectID}}) +} + +func sessionID(r *http.Request) domain.SessionID { + return domain.SessionID(chi.URLParam(r, "sessionId")) +} + +func parseSessionListFilter(r *http.Request) (service.SessionListFilter, error) { + q := r.URL.Query() + filter := service.SessionListFilter{ProjectID: domain.ProjectID(q.Get("project"))} + if raw := q.Get("active"); raw != "" { + active, err := strconv.ParseBool(raw) + if err != nil { + return service.SessionListFilter{}, errors.New("active must be a boolean") + } + filter.Active = &active + } + if raw := q.Get("orchestratorOnly"); raw != "" { + orchestratorOnly, err := strconv.ParseBool(raw) + if err != nil { + return service.SessionListFilter{}, errors.New("orchestratorOnly must be a boolean") + } + filter.OrchestratorOnly = orchestratorOnly + } + if raw := q.Get("fresh"); raw != "" { + fresh, err := strconv.ParseBool(raw) + if err != nil { + return service.SessionListFilter{}, errors.New("fresh must be a boolean") + } + filter.Fresh = fresh + } + return filter, nil +} + +func stripUnsafeControlChars(message string) string { + return strings.Map(func(r rune) rune { + if unicode.IsControl(r) && r != '\n' && r != '\r' && r != '\t' { + return -1 + } + return r + }, message) +} + +func writeSessionError(w http.ResponseWriter, r *http.Request, err error) { + switch { + case errors.Is(err, sessionmanager.ErrNotFound): + envelope.WriteAPIError(w, r, http.StatusNotFound, "not_found", "SESSION_NOT_FOUND", "Unknown session", nil) + case errors.Is(err, sessionmanager.ErrNotRestorable): + envelope.WriteAPIError(w, r, http.StatusConflict, "conflict", "SESSION_NOT_RESTORABLE", "Session is not restorable", nil) + case errors.Is(err, sessionmanager.ErrIncompleteHandle): + envelope.WriteAPIError(w, r, http.StatusConflict, "conflict", "SESSION_INCOMPLETE_HANDLE", "Session is missing runtime or workspace handles", nil) + default: + envelope.WriteAPIError(w, r, http.StatusInternalServerError, "internal", "SESSION_OPERATION_FAILED", "Session operation failed", nil) + } +} diff --git a/backend/internal/httpd/controllers/sessions_test.go b/backend/internal/httpd/controllers/sessions_test.go new file mode 100644 index 000000000..62ef4c8c3 --- /dev/null +++ b/backend/internal/httpd/controllers/sessions_test.go @@ -0,0 +1,174 @@ +package controllers_test + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/config" + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/httpd" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" + "github.com/aoagents/agent-orchestrator/backend/internal/service" +) + +type fakeSessionService struct { + sessions map[domain.SessionID]domain.Session + sent string +} + +func newFakeSessionService() *fakeSessionService { + now := time.Now().UTC() + s := domain.Session{SessionRecord: domain.SessionRecord{ID: "ao-1", ProjectID: "ao", Kind: domain.KindWorker, Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: now}, CreatedAt: now, UpdatedAt: now}, Status: domain.StatusIdle} + return &fakeSessionService{sessions: map[domain.SessionID]domain.Session{s.ID: s}} +} + +func (f *fakeSessionService) List(_ context.Context, filter service.SessionListFilter) ([]domain.Session, error) { + var out []domain.Session + for _, s := range f.sessions { + if filter.ProjectID != "" && s.ProjectID != filter.ProjectID { + continue + } + if filter.Active != nil && s.IsTerminated == *filter.Active { + continue + } + if filter.OrchestratorOnly && s.Kind != domain.KindOrchestrator { + continue + } + out = append(out, s) + } + return out, nil +} + +func (f *fakeSessionService) Spawn(_ context.Context, cfg ports.SpawnConfig) (domain.Session, error) { + now := time.Now().UTC() + s := domain.Session{SessionRecord: domain.SessionRecord{ID: domain.SessionID(string(cfg.ProjectID) + "-2"), ProjectID: cfg.ProjectID, IssueID: cfg.IssueID, Kind: cfg.Kind, Harness: cfg.Harness, Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: now}, CreatedAt: now, UpdatedAt: now}, Status: domain.StatusIdle} + f.sessions[s.ID] = s + return s, nil +} + +func (f *fakeSessionService) Get(_ context.Context, id domain.SessionID) (domain.Session, error) { + return f.sessions[id], nil +} + +func (f *fakeSessionService) Restore(_ context.Context, id domain.SessionID) (domain.Session, error) { + s := f.sessions[id] + s.IsTerminated = false + s.Status = domain.StatusIdle + f.sessions[id] = s + return s, nil +} + +func (f *fakeSessionService) Kill(_ context.Context, id domain.SessionID) (bool, error) { + s := f.sessions[id] + s.IsTerminated = true + s.Status = domain.StatusTerminated + f.sessions[id] = s + return true, nil +} + +func (f *fakeSessionService) Send(_ context.Context, _ domain.SessionID, message string) error { + f.sent = message + return nil +} + +func newSessionTestServer(t *testing.T, svc *fakeSessionService) *httptest.Server { + t.Helper() + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + srv := httptest.NewServer(httpd.NewRouterWithAPI(config.Config{}, log, nil, httpd.APIDeps{Sessions: svc})) + t.Cleanup(srv.Close) + return srv +} + +func TestSessionsRoutes_DefaultToStubsWithoutService(t *testing.T) { + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + srv := httptest.NewServer(httpd.NewRouter(config.Config{}, log, nil)) + t.Cleanup(srv.Close) + + body, status, headers := doRequest(t, srv, "GET", "/api/v1/sessions", "") + assertJSON(t, headers) + assertErrorCode(t, body, status, http.StatusNotImplemented, "NOT_IMPLEMENTED") +} + +func TestSessionsAPI_ListSpawnGetAndActions(t *testing.T) { + svc := newFakeSessionService() + srv := newSessionTestServer(t, svc) + + body, status, _ := doRequest(t, srv, "GET", "/api/v1/sessions?project=ao", "") + if status != http.StatusOK { + t.Fatalf("GET sessions = %d, want 200; body=%s", status, body) + } + var list struct { + Sessions []sessionBody `json:"sessions"` + } + mustJSON(t, body, &list) + if len(list.Sessions) != 1 || list.Sessions[0].ID != "ao-1" || list.Sessions[0].Status != string(domain.StatusIdle) { + t.Fatalf("list = %#v", list) + } + + body, status, _ = doRequest(t, srv, "POST", "/api/v1/sessions", `{"projectId":"ao","issueId":"ISS-1","kind":"worker","harness":"codex","prompt":"fix"}`) + if status != http.StatusCreated { + t.Fatalf("POST session = %d, want 201; body=%s", status, body) + } + var spawned struct { + Session sessionBody `json:"session"` + } + mustJSON(t, body, &spawned) + if spawned.Session.ID != "ao-2" || spawned.Session.IssueID != "ISS-1" || spawned.Session.Harness != "codex" { + t.Fatalf("spawned = %#v", spawned) + } + + body, status, _ = doRequest(t, srv, "GET", "/api/v1/sessions/ao-2", "") + if status != http.StatusOK { + t.Fatalf("GET session = %d, want 200; body=%s", status, body) + } + + body, status, _ = doRequest(t, srv, "POST", "/api/v1/sessions/ao-2/send", "{\"message\":\"con\\u0000tinue\"}") + if status != http.StatusOK || svc.sent != "continue" { + t.Fatalf("send status=%d sent=%q body=%s", status, svc.sent, body) + } + + body, status, _ = doRequest(t, srv, "POST", "/api/v1/sessions/ao-2/kill", "") + if status != http.StatusOK { + t.Fatalf("kill = %d, want 200; body=%s", status, body) + } + var killed struct { + SessionID string `json:"sessionId"` + Freed bool `json:"freed"` + } + mustJSON(t, body, &killed) + if killed.SessionID != "ao-2" || !killed.Freed { + t.Fatalf("kill response = %#v", killed) + } + + body, status, _ = doRequest(t, srv, "POST", "/api/v1/sessions/ao-2/restore", "") + if status != http.StatusOK { + t.Fatalf("restore = %d, want 200; body=%s", status, body) + } + + body, status, _ = doRequest(t, srv, "PATCH", "/api/v1/sessions/ao-2", `{"displayName":"Renamed"}`) + assertErrorCode(t, body, status, http.StatusNotImplemented, "NOT_IMPLEMENTED") + + body, status, _ = doRequest(t, srv, "POST", "/api/v1/orchestrators", `{"projectId":"ao"}`) + if status != http.StatusCreated { + t.Fatalf("orchestrator = %d, want 201; body=%s", status, body) + } +} + +func TestSessionsAPI_SendValidation(t *testing.T) { + srv := newSessionTestServer(t, newFakeSessionService()) + + body, status, _ := doRequest(t, srv, "POST", "/api/v1/sessions/ao-1/send", `{"message":""}`) + assertErrorCode(t, body, status, http.StatusBadRequest, "MESSAGE_REQUIRED") +} + +type sessionBody struct { + ID string `json:"id"` + IssueID string `json:"issueId"` + Harness string `json:"harness"` + Status string `json:"status"` +} diff --git a/backend/internal/integration/lifecycle_sqlite_test.go b/backend/internal/integration/lifecycle_sqlite_test.go index 670fa150a..b8d363328 100644 --- a/backend/internal/integration/lifecycle_sqlite_test.go +++ b/backend/internal/integration/lifecycle_sqlite_test.go @@ -11,7 +11,8 @@ import ( "github.com/aoagents/agent-orchestrator/backend/internal/ports" prsvc "github.com/aoagents/agent-orchestrator/backend/internal/pr" "github.com/aoagents/agent-orchestrator/backend/internal/project" - "github.com/aoagents/agent-orchestrator/backend/internal/session" + "github.com/aoagents/agent-orchestrator/backend/internal/service" + sessionmanager "github.com/aoagents/agent-orchestrator/backend/internal/session_manager" "github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite" ) @@ -54,7 +55,7 @@ func (c *captureMessenger) Send(_ context.Context, _ domain.SessionID, msg strin type stack struct { store *sqlite.Store - sm *session.Manager + sm *service.Session lcm *lifecycle.Manager prm *prsvc.Manager rt *stubRuntime @@ -78,7 +79,8 @@ func newStack(t *testing.T) *stack { prm := prsvc.New(prsvc.Deps{Writer: store, Lifecycle: lcm}) rt := &stubRuntime{} ws := &stubWorkspace{} - sm := session.New(session.Deps{Runtime: rt, Agent: stubAgent{}, Workspace: ws, Store: store, Messenger: msg, Lifecycle: lcm}) + mgr := sessionmanager.New(sessionmanager.Deps{Runtime: rt, Agent: stubAgent{}, Workspace: ws, Store: store, Messenger: msg, Lifecycle: lcm}) + sm := service.NewSession(mgr, store) return &stack{store: store, sm: sm, lcm: lcm, prm: prm, rt: rt, ws: ws, msg: msg} } diff --git a/backend/internal/service/session.go b/backend/internal/service/session.go new file mode 100644 index 000000000..fae8d76bd --- /dev/null +++ b/backend/internal/service/session.go @@ -0,0 +1,143 @@ +package service + +import ( + "context" + "fmt" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" + sessionmanager "github.com/aoagents/agent-orchestrator/backend/internal/session_manager" +) + +// SessionStore is the read-only persistence surface needed to assemble controller-facing session read models. +type SessionStore interface { + GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) + ListSessions(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) + ListAllSessions(ctx context.Context) ([]domain.SessionRecord, error) + GetDisplayPRFactsForSession(ctx context.Context, id domain.SessionID) (domain.PRFacts, bool, error) +} + +// SessionListFilter captures API-facing session list query filters. +type SessionListFilter struct { + ProjectID domain.ProjectID + Active *bool + OrchestratorOnly bool + Fresh bool +} + +// Session is the controller-facing session service. It delegates command-side +// session operations to the internal sessionmanager.Manager and owns read-model +// assembly, including user-facing display status derivation. +type Session struct { + manager *sessionmanager.Manager + store SessionStore +} + +// NewSession wires a controller-facing session service over an internal session Manager. +func NewSession(manager *sessionmanager.Manager, store SessionStore) *Session { + return &Session{manager: manager, store: store} +} + +// Spawn creates a session and returns the API-facing read model. +func (s *Session) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Session, error) { + rec, err := s.manager.Spawn(ctx, cfg) + if err != nil { + return domain.Session{}, err + } + return s.toSession(ctx, rec) +} + +// Restore relaunches a terminated session and returns the API-facing read model. +func (s *Session) Restore(ctx context.Context, id domain.SessionID) (domain.Session, error) { + rec, err := s.manager.Restore(ctx, id) + if err != nil { + return domain.Session{}, err + } + return s.toSession(ctx, rec) +} + +// Kill delegates terminal intent and teardown to the internal manager. +func (s *Session) Kill(ctx context.Context, id domain.SessionID) (bool, error) { + return s.manager.Kill(ctx, id) +} + +// Send delegates agent messaging to the internal manager. +func (s *Session) Send(ctx context.Context, id domain.SessionID, message string) error { + return s.manager.Send(ctx, id, message) +} + +// Cleanup delegates terminal workspace cleanup to the internal manager. +func (s *Session) Cleanup(ctx context.Context, project domain.ProjectID) ([]domain.SessionID, error) { + return s.manager.Cleanup(ctx, project) +} + +// List returns sessions as enriched display models after applying API filters. +func (s *Session) List(ctx context.Context, filter SessionListFilter) ([]domain.Session, error) { + recs, err := s.listRecords(ctx, filter.ProjectID) + if err != nil { + return nil, err + } + out := make([]domain.Session, 0, len(recs)) + for _, rec := range recs { + if !matchesSessionFilter(rec, filter) { + continue + } + sess, err := s.toSession(ctx, rec) + if err != nil { + return nil, err + } + out = append(out, sess) + } + return out, nil +} + +func (s *Session) listRecords(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) { + if project == "" { + recs, err := s.store.ListAllSessions(ctx) + if err != nil { + return nil, fmt.Errorf("list all sessions: %w", err) + } + return recs, nil + } + recs, err := s.store.ListSessions(ctx, project) + if err != nil { + return nil, fmt.Errorf("list %s: %w", project, err) + } + return recs, nil +} + +func matchesSessionFilter(rec domain.SessionRecord, filter SessionListFilter) bool { + if filter.Active != nil && rec.IsTerminated == *filter.Active { + return false + } + if filter.OrchestratorOnly && rec.Kind != domain.KindOrchestrator { + return false + } + if filter.Fresh && rec.IsTerminated { + return false + } + return true +} + +// Get returns one session as an enriched display model, or sessionmanager.ErrNotFound if it is absent. +func (s *Session) Get(ctx context.Context, id domain.SessionID) (domain.Session, error) { + rec, ok, err := s.store.GetSession(ctx, id) + if err != nil { + return domain.Session{}, fmt.Errorf("get %s: %w", id, err) + } + if !ok { + return domain.Session{}, fmt.Errorf("get %s: %w", id, sessionmanager.ErrNotFound) + } + return s.toSession(ctx, rec) +} + +func (s *Session) toSession(ctx context.Context, rec domain.SessionRecord) (domain.Session, error) { + pr, ok, err := s.store.GetDisplayPRFactsForSession(ctx, rec.ID) + if err != nil { + return domain.Session{}, fmt.Errorf("pr facts %s: %w", rec.ID, err) + } + if !ok { + return domain.Session{SessionRecord: rec, Status: deriveStatus(rec, nil)}, nil + } + return domain.Session{SessionRecord: rec, Status: deriveStatus(rec, &pr)}, nil +} diff --git a/backend/internal/service/session_status.go b/backend/internal/service/session_status.go new file mode 100644 index 000000000..801b7b191 --- /dev/null +++ b/backend/internal/service/session_status.go @@ -0,0 +1,49 @@ +package service + +import "github.com/aoagents/agent-orchestrator/backend/internal/domain" + +func deriveStatus(rec domain.SessionRecord, pr *domain.PRFacts) domain.SessionStatus { + if rec.IsTerminated { + if pr != nil && pr.Merged { + return domain.StatusMerged + } + return domain.StatusTerminated + } + + if rec.Activity.State == domain.ActivityWaitingInput { + return domain.StatusNeedsInput + } + + if pr != nil { + if pr.Merged { + return domain.StatusMerged + } + if !pr.Closed { + return prPipelineStatus(*pr) + } + } + + if rec.Activity.State == domain.ActivityActive { + return domain.StatusWorking + } + return domain.StatusIdle +} + +func prPipelineStatus(pr domain.PRFacts) domain.SessionStatus { + switch { + case pr.CI == domain.CIFailing: + return domain.StatusCIFailed + case pr.Draft: + return domain.StatusDraft + case pr.Review == domain.ReviewChangesRequest || pr.ReviewComments: + return domain.StatusChangesRequested + case pr.Mergeability == domain.MergeMergeable: + return domain.StatusMergeable + case pr.Review == domain.ReviewApproved: + return domain.StatusApproved + case pr.Review == domain.ReviewRequired: + return domain.StatusReviewPending + default: + return domain.StatusPROpen + } +} diff --git a/backend/internal/service/session_status_test.go b/backend/internal/service/session_status_test.go new file mode 100644 index 000000000..b45125e12 --- /dev/null +++ b/backend/internal/service/session_status_test.go @@ -0,0 +1,42 @@ +package service + +import ( + "testing" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +func statusRec(activity domain.ActivityState, terminated bool) domain.SessionRecord { + return domain.SessionRecord{Activity: domain.Activity{State: activity}, IsTerminated: terminated} +} + +func statusPR(facts domain.PRFacts) *domain.PRFacts { return &facts } + +func TestServiceDerivesStatusFromSessionFactsAndPR(t *testing.T) { + tests := []struct { + name string + rec domain.SessionRecord + pr *domain.PRFacts + want domain.SessionStatus + }{ + {"terminated", statusRec(domain.ActivityExited, true), nil, domain.StatusTerminated}, + {"merged-pr", statusRec(domain.ActivityIdle, true), statusPR(domain.PRFacts{Merged: true}), domain.StatusMerged}, + {"needs-input", statusRec(domain.ActivityWaitingInput, false), statusPR(domain.PRFacts{CI: domain.CIFailing}), domain.StatusNeedsInput}, + {"ci-failed", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{CI: domain.CIFailing}), domain.StatusCIFailed}, + {"draft", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{Draft: true}), domain.StatusDraft}, + {"changes-requested", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{Review: domain.ReviewChangesRequest}), domain.StatusChangesRequested}, + {"mergeable", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{Mergeability: domain.MergeMergeable}), domain.StatusMergeable}, + {"approved", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{Review: domain.ReviewApproved}), domain.StatusApproved}, + {"review-pending", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{Review: domain.ReviewRequired}), domain.StatusReviewPending}, + {"pr-open", statusRec(domain.ActivityIdle, false), statusPR(domain.PRFacts{}), domain.StatusPROpen}, + {"working", statusRec(domain.ActivityActive, false), nil, domain.StatusWorking}, + {"idle", statusRec(domain.ActivityIdle, false), nil, domain.StatusIdle}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := deriveStatus(tt.rec, tt.pr); got != tt.want { + t.Fatalf("got %q want %q", got, tt.want) + } + }) + } +} diff --git a/backend/internal/service/session_test.go b/backend/internal/service/session_test.go new file mode 100644 index 000000000..f092dc195 --- /dev/null +++ b/backend/internal/service/session_test.go @@ -0,0 +1,68 @@ +package service + +import ( + "context" + "fmt" + "testing" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" +) + +type fakeSessionStore struct { + sessions map[domain.SessionID]domain.SessionRecord + pr map[domain.SessionID]domain.PRFacts + num int +} + +func newFakeSessionStore() *fakeSessionStore { + return &fakeSessionStore{sessions: map[domain.SessionID]domain.SessionRecord{}, pr: map[domain.SessionID]domain.PRFacts{}} +} + +func (f *fakeSessionStore) CreateSession(_ context.Context, rec domain.SessionRecord) (domain.SessionRecord, error) { + f.num++ + rec.ID = domain.SessionID(fmt.Sprintf("%s-%d", rec.ProjectID, f.num)) + f.sessions[rec.ID] = rec + return rec, nil +} + +func (f *fakeSessionStore) GetSession(_ context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) { + r, ok := f.sessions[id] + return r, ok, nil +} + +func (f *fakeSessionStore) ListSessions(_ context.Context, p domain.ProjectID) ([]domain.SessionRecord, error) { + var out []domain.SessionRecord + for _, r := range f.sessions { + if r.ProjectID == p { + out = append(out, r) + } + } + return out, nil +} + +func (f *fakeSessionStore) ListAllSessions(_ context.Context) ([]domain.SessionRecord, error) { + out := make([]domain.SessionRecord, 0, len(f.sessions)) + for _, r := range f.sessions { + out = append(out, r) + } + return out, nil +} + +func (f *fakeSessionStore) GetDisplayPRFactsForSession(_ context.Context, id domain.SessionID) (domain.PRFacts, bool, error) { + pr, ok := f.pr[id] + return pr, ok, nil +} + +func TestSessionListDerivesStatusFromPRFacts(t *testing.T) { + st := newFakeSessionStore() + st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityActive}} + st.pr["mer-1"] = domain.PRFacts{URL: "pr1", CI: domain.CIFailing} + + list, err := (&Session{store: st}).List(context.Background(), SessionListFilter{ProjectID: "mer"}) + if err != nil { + t.Fatal(err) + } + if len(list) != 1 || list[0].Status != domain.StatusCIFailed { + t.Fatalf("got %+v", list) + } +} diff --git a/backend/internal/session/manager.go b/backend/internal/session_manager/manager.go similarity index 76% rename from backend/internal/session/manager.go rename to backend/internal/session_manager/manager.go index ca4d0fa68..18c21c5c4 100644 --- a/backend/internal/session/manager.go +++ b/backend/internal/session_manager/manager.go @@ -1,7 +1,6 @@ -// Package session drives the runtime/agent/workspace plugins to create and tear -// down sessions, routes durable lifecycle fact writes through lifecycle, and -// attaches derived display status on read. -package session +// Package sessionmanager drives internal session command operations over runtime, +// agent, workspace, storage, messenger, and lifecycle dependencies. +package sessionmanager import ( "context" @@ -37,20 +36,20 @@ type runtimeController interface { Destroy(ctx context.Context, handle ports.RuntimeHandle) error } -type sessionStore interface { +// Store is the persistence surface needed by the internal session Manager. +type Store interface { CreateSession(ctx context.Context, rec domain.SessionRecord) (domain.SessionRecord, error) GetSession(ctx context.Context, id domain.SessionID) (domain.SessionRecord, bool, error) ListSessions(ctx context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) - GetDisplayPRFactsForSession(ctx context.Context, id domain.SessionID) (domain.PRFacts, bool, error) } -// Manager coordinates session spawn, restore, kill, listing, and cleanup over -// the outbound ports. +// Manager coordinates internal session spawn, restore, kill, and cleanup over +// the outbound ports. User-facing read-model assembly lives in the service package. type Manager struct { runtime runtimeController agent ports.Agent workspace ports.Workspace - store sessionStore + store Store messenger ports.AgentMessenger lcm lifecycleRecorder clock func() time.Time @@ -61,7 +60,7 @@ type Deps struct { Runtime runtimeController Agent ports.Agent Workspace ports.Workspace - Store sessionStore + Store Store Messenger ports.AgentMessenger Lifecycle lifecycleRecorder Clock func() time.Time @@ -88,17 +87,17 @@ func New(d Deps) *Manager { // Spawn creates the session row (which assigns the "{project}-{n}" id), then the // workspace and runtime, then reports completion to the LCM. A failure after the // row exists parks it as terminated and rolls back what was built. -func (m *Manager) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Session, error) { +func (m *Manager) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.SessionRecord, error) { rec, err := m.store.CreateSession(ctx, seedRecord(cfg, m.clock())) if err != nil { - return domain.Session{}, fmt.Errorf("spawn: create: %w", err) + return domain.SessionRecord{}, fmt.Errorf("spawn: create: %w", err) } id := rec.ID ws, err := m.workspace.Create(ctx, ports.WorkspaceConfig{ProjectID: cfg.ProjectID, SessionID: id, Branch: cfg.Branch}) if err != nil { m.markSpawnFailedTerminated(ctx, id) - return domain.Session{}, fmt.Errorf("spawn %s: workspace: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("spawn %s: workspace: %w", id, err) } agentCfg := ports.AgentConfig{SessionID: id, WorkspacePath: ws.Path, Prompt: buildPrompt(cfg)} @@ -111,7 +110,7 @@ func (m *Manager) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Sess if err != nil { _ = m.workspace.Destroy(ctx, ws) m.markSpawnFailedTerminated(ctx, id) - return domain.Session{}, fmt.Errorf("spawn %s: runtime: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("spawn %s: runtime: %w", id, err) } metadata := domain.SessionMetadata{Branch: ws.Branch, WorkspacePath: ws.Path, RuntimeHandleID: handle.ID, Prompt: agentCfg.Prompt} @@ -119,9 +118,9 @@ func (m *Manager) Spawn(ctx context.Context, cfg ports.SpawnConfig) (domain.Sess _ = m.runtime.Destroy(ctx, handle) _ = m.workspace.Destroy(ctx, ws) m.markSpawnFailedTerminated(ctx, id) - return domain.Session{}, fmt.Errorf("spawn %s: completed: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("spawn %s: completed: %w", id, err) } - return m.Get(ctx, id) + return m.getRecord(ctx, id) } // markSpawnFailedTerminated best-effort parks an orphaned spawn as terminated. @@ -161,25 +160,25 @@ func (m *Manager) Kill(ctx context.Context, id domain.SessionID) (bool, error) { // Restore relaunches a torn-down session in its workspace. The fallible I/O runs // before any durable session write, so a failure never resurrects the row or destroys // the worktree (it may hold the agent's prior work). -func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.Session, error) { +func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.SessionRecord, error) { rec, ok, err := m.store.GetSession(ctx, id) if err != nil { - return domain.Session{}, fmt.Errorf("restore %s: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("restore %s: %w", id, err) } if !ok { - return domain.Session{}, fmt.Errorf("restore %s: %w", id, ErrNotFound) + return domain.SessionRecord{}, fmt.Errorf("restore %s: %w", id, ErrNotFound) } if !rec.IsTerminated { - return domain.Session{}, fmt.Errorf("restore %s: %w", id, ErrNotRestorable) + return domain.SessionRecord{}, fmt.Errorf("restore %s: %w", id, ErrNotRestorable) } meta := rec.Metadata if meta.AgentSessionID == "" && meta.Prompt == "" { - return domain.Session{}, fmt.Errorf("restore %s: nothing to resume from", id) + return domain.SessionRecord{}, fmt.Errorf("restore %s: nothing to resume from", id) } ws, err := m.workspace.Restore(ctx, ports.WorkspaceConfig{ProjectID: rec.ProjectID, SessionID: id, Branch: meta.Branch}) if err != nil { - return domain.Session{}, fmt.Errorf("restore %s: workspace: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("restore %s: workspace: %w", id, err) } agentCfg := ports.AgentConfig{SessionID: id, WorkspacePath: ws.Path, Prompt: meta.Prompt} launch := m.agent.GetRestoreCommand(meta.AgentSessionID) @@ -193,43 +192,25 @@ func (m *Manager) Restore(ctx context.Context, id domain.SessionID) (domain.Sess Env: spawnEnv(m.agent.GetEnvironment(agentCfg), id, rec.ProjectID, rec.IssueID), }) if err != nil { - return domain.Session{}, fmt.Errorf("restore %s: runtime: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("restore %s: runtime: %w", id, err) } metadata := domain.SessionMetadata{Branch: ws.Branch, WorkspacePath: ws.Path, RuntimeHandleID: handle.ID, AgentSessionID: meta.AgentSessionID, Prompt: meta.Prompt} if err := m.lcm.MarkSpawned(ctx, id, metadata); err != nil { _ = m.runtime.Destroy(ctx, handle) - return domain.Session{}, fmt.Errorf("restore %s: completed: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("restore %s: completed: %w", id, err) } - return m.Get(ctx, id) + return m.getRecord(ctx, id) } -// List returns the project's sessions as enriched display models. -func (m *Manager) List(ctx context.Context, project domain.ProjectID) ([]domain.Session, error) { - recs, err := m.store.ListSessions(ctx, project) - if err != nil { - return nil, fmt.Errorf("list %s: %w", project, err) - } - out := make([]domain.Session, 0, len(recs)) - for _, rec := range recs { - s, err := m.toSession(ctx, rec) - if err != nil { - return nil, err - } - out = append(out, s) - } - return out, nil -} - -// Get returns one session as a display model, or ErrNotFound if it is absent. -func (m *Manager) Get(ctx context.Context, id domain.SessionID) (domain.Session, error) { +func (m *Manager) getRecord(ctx context.Context, id domain.SessionID) (domain.SessionRecord, error) { rec, ok, err := m.store.GetSession(ctx, id) if err != nil { - return domain.Session{}, fmt.Errorf("get %s: %w", id, err) + return domain.SessionRecord{}, fmt.Errorf("get %s: %w", id, err) } if !ok { - return domain.Session{}, fmt.Errorf("get %s: %w", id, ErrNotFound) + return domain.SessionRecord{}, fmt.Errorf("get %s: %w", id, ErrNotFound) } - return m.toSession(ctx, rec) + return rec, nil } // Send delivers a message to a running session's agent via the messenger. @@ -269,17 +250,6 @@ func (m *Manager) Cleanup(ctx context.Context, project domain.ProjectID) ([]doma // ---- helpers ---- -func (m *Manager) toSession(ctx context.Context, rec domain.SessionRecord) (domain.Session, error) { - pr, ok, err := m.store.GetDisplayPRFactsForSession(ctx, rec.ID) - if err != nil { - return domain.Session{}, fmt.Errorf("pr facts %s: %w", rec.ID, err) - } - if !ok { - return domain.Session{SessionRecord: rec, Status: domain.DeriveStatus(rec, nil)}, nil - } - return domain.Session{SessionRecord: rec, Status: domain.DeriveStatus(rec, &pr)}, nil -} - func seedRecord(cfg ports.SpawnConfig, now time.Time) domain.SessionRecord { return domain.SessionRecord{ ProjectID: cfg.ProjectID, diff --git a/backend/internal/session/manager_test.go b/backend/internal/session_manager/manager_test.go similarity index 92% rename from backend/internal/session/manager_test.go rename to backend/internal/session_manager/manager_test.go index f682a51af..22ef78755 100644 --- a/backend/internal/session/manager_test.go +++ b/backend/internal/session_manager/manager_test.go @@ -1,4 +1,4 @@ -package session +package sessionmanager import ( "context" @@ -149,8 +149,8 @@ func TestSpawn_AssignsIDAndGoesIdle(t *testing.T) { if s.ID != "mer-1" { t.Fatalf("got %q", s.ID) } - if s.Status != domain.StatusIdle { - t.Fatalf("fresh session displays idle, got %q", s.Status) + if s.Activity.State != domain.ActivityIdle { + t.Fatalf("fresh session records idle, got %q", s.Activity.State) } if rt.created != 1 { t.Fatal("runtime not created") @@ -197,8 +197,8 @@ func TestRestore_ReopensTerminal(t *testing.T) { if err != nil { t.Fatal(err) } - if s.Status != domain.StatusIdle { - t.Fatalf("restored displays idle, got %q", s.Status) + if s.Activity.State != domain.ActivityIdle { + t.Fatalf("restored records idle, got %q", s.Activity.State) } if rt.created != 1 { t.Fatal("restore should relaunch") @@ -211,18 +211,6 @@ func TestRestore_RefusesLiveSession(t *testing.T) { t.Fatalf("want ErrNotRestorable, got %v", err) } } -func TestList_DerivesStatusFromPRFacts(t *testing.T) { - m, st, _, _ := newManager() - st.sessions["mer-1"] = mkLive("mer-1") - st.pr["mer-1"] = domain.PRFacts{URL: "pr1", CI: domain.CIFailing} - list, err := m.List(ctx, "mer") - if err != nil { - t.Fatal(err) - } - if len(list) != 1 || list[0].Status != domain.StatusCIFailed { - t.Fatalf("got %+v", list) - } -} func TestCleanup_ReclaimsTerminalWorkspaces(t *testing.T) { m, st, _, ws := newManager() seedTerminal(st, "mer-1", domain.SessionMetadata{WorkspacePath: "/ws/mer-1"}) diff --git a/docs/README.md b/docs/README.md index ad4c1453e..e14cb669c 100644 --- a/docs/README.md +++ b/docs/README.md @@ -14,4 +14,4 @@ Persist durable facts, derive display status: - session table: `activity_state`, `is_terminated`, identity, metadata - PR tables: PR/CI/review facts -- derived read model: `domain.DeriveStatus(session, prFacts)` +- derived read model: `service.Session` computes display status from session + PR facts diff --git a/docs/architecture.md b/docs/architecture.md index 8e768c55f..c3c26affa 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -17,15 +17,16 @@ The durable session facts are: - `is_terminated` — whether the session should be treated as over. - PR facts in the `pr`, `pr_checks`, and `pr_comment` tables. -The UI status is not stored. `domain.DeriveStatus` computes it from the session -record plus PR facts. +The UI status is not stored. `service.Session` computes it from the session +record plus PR facts while assembling controller-facing read models. ## Package layout ``` -backend/internal/domain shared vocabulary and display-status derivation +backend/internal/domain shared vocabulary and API status value types backend/internal/ports inbound/outbound interfaces -backend/internal/session explicit mutations: spawn, kill, restore, send, cleanup +backend/internal/service controller-facing services and read-model assembly +backend/internal/session_manager internal session command manager backend/internal/lifecycle runtime/activity/spawn/termination session fact reducer backend/internal/pr PR observation ingestion backend/internal/storage SQLite persistence and DB-triggered CDC @@ -37,8 +38,8 @@ backend/internal/adapters Zellij/git-worktree/GitHub adapters ## Status derivation -`session.Manager` selects the display PR from all PR snapshots for a session, then -`domain.DeriveStatus(session, prFacts)` applies this rough precedence: +`service.Session` selects the display PR from all PR snapshots for a session, then +applies this rough precedence: 1. `is_terminated` → `terminated`, except merged PRs display `merged`. 2. `activity_state=waiting_input` → `needs_input`. @@ -69,14 +70,16 @@ consumed at read time for display status. ## Session manager -`session.Manager` performs explicit user mutations: +`session_manager.Manager` performs internal session mutations: - `Spawn` creates a row, creates workspace/runtime resources, and reports the handles to the lifecycle manager. - `Kill` marks the row terminated, then tears down runtime/workspace resources. - `Restore` relaunches a terminated session and clears `is_terminated` via the spawn-completed path. -- `List`/`Get` attach the derived display status. + +`service.Session` is the controller-facing boundary. It delegates commands to +`session_manager.Manager` and attaches derived display status on read paths. ## Persistence and CDC diff --git a/docs/status.md b/docs/status.md index 6ca5bc274..1da4946a3 100644 --- a/docs/status.md +++ b/docs/status.md @@ -1,7 +1,7 @@ # agent-orchestrator status Current main contains the Go backend daemon, Cobra CLI foundation, SQLite store, -CDC poller/broadcaster, lifecycle/session managers, terminal mux, project API +CDC poller/broadcaster, lifecycle/session services, terminal mux, project API controller/manager work, runtime/workspace/tracker adapters, and CDC-backed event rows. ## Build & test @@ -18,9 +18,11 @@ npm run lint from those plus PR facts. - SQLite: migrations create projects, sessions, PR/check/comment, and `change_log` tables. - CDC: DB triggers append to `change_log`; the poller broadcasts live events. -- Session Manager: spawn/kill/restore/list/get/send/cleanup over runtime, - workspace, agent, store, messenger, and lifecycle ports. It is package-level - code today; daemon HTTP routes for session commands are not wired yet. +- Session Manager: internal spawn/kill/restore/send/cleanup over runtime, + workspace, agent, store, messenger, and lifecycle ports. +- Service package: controller-facing session boundary that delegates commands to + the manager and assembles list/get/spawn/restore read models with display status. + Daemon HTTP routes for session commands are not wired yet. ## Next integration work