Co-authored-by: itrytoohard <ayetrytoohard@gmail.com>
This commit is contained in:
prateek
GitHub
parent
f9b08aada4
commit
424e6e824b
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
@ -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"`
|
||||
}
|
||||
|
|
@ -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}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
@ -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,
|
||||
|
|
@ -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"})
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue