bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

refactor: remove activity source tracking (#62) (#66) 

Co-authored-by: itrytoohard <ayetrytoohard@gmail.com>
This commit is contained in:
prateek 2026-06-01 09:26:18 +05:30 committed by GitHub
parent a34094e7d8
commit c8f6050539
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
22 changed files with 70 additions and 148 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/internal/cdc/cdc_test.go
View File

@ -32,7 +32,7 @@ func seedSession(t *testing.T, s *sqlite.Store) domain.SessionRecord {
}
r, err := s.CreateSession(ctx, domain.SessionRecord{
ProjectID: "mer", Kind: domain.KindWorker,
Activity: domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: now, Source: domain.SourceNative},
Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: now},
CreatedAt: now, UpdatedAt: now,
})
if err != nil {

4
backend/internal/daemon/wiring_test.go
View File

@ -42,14 +42,14 @@ func TestWiring_WriteFlowsToBroadcaster(t *testing.T) {
}
rec, err := store.CreateSession(ctx, domain.SessionRecord{
ProjectID: "mer", Kind: domain.KindWorker,
Activity: domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: time.Now(), Source: domain.SourceNone},
Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: time.Now()},
})
if err != nil {
t.Fatal(err)
}
// A real transition through the engine, which writes the row and fires the
// activity_state/is_terminated CDC trigger.
if err := lcm.ApplyActivitySignal(ctx, rec.ID, ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Timestamp: time.Now(), Source: domain.SourceNative}); err != nil {
if err := lcm.ApplyActivitySignal(ctx, rec.ID, ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Timestamp: time.Now()}); err != nil {
t.Fatal(err)
}

46
backend/internal/domain/activity.go
View File

@ -5,59 +5,23 @@ import "time"
// ActivityState is how busy the agent is, derived from its output/JSONL.
type ActivityState string
// Activity states. WaitingInput and Blocked are sticky (see IsSticky).
// Activity states. WaitingInput is sticky (see IsSticky).
const (
ActivityActive ActivityState = "active"
ActivityIdle ActivityState = "idle"
ActivityWaitingInput ActivityState = "waiting_input"
ActivityBlocked ActivityState = "blocked"
ActivityExited ActivityState = "exited"
)
// IsSticky reports whether an activity state must NOT be aged/demoted by the
// passage of time (a paused agent is still paused until a new signal says so).
func (a ActivityState) IsSticky() bool {
return a == ActivityWaitingInput || a == ActivityBlocked
return a == ActivityWaitingInput
}
// ActivitySource records where an activity reading came from, so a weaker
// source can't override a stronger one.
type ActivitySource string
// Activity signal sources, strongest first.
const (
SourceNative ActivitySource = "native"
SourceTerminal ActivitySource = "terminal"
SourceHook ActivitySource = "hook"
SourceRuntime ActivitySource = "runtime"
SourceNone ActivitySource = "none"
)
// CanOverride reports whether a reading from source a may replace a current
// reading from source current. Unknown sources are treated as weakest.
func (a ActivitySource) CanOverride(current ActivitySource) bool {
return activitySourceRank(a) <= activitySourceRank(current)
}
func activitySourceRank(s ActivitySource) int {
switch s {
case SourceNative:
return 0
case SourceTerminal:
return 1
case SourceHook:
return 2
case SourceRuntime:
return 3
default:
return 4
}
}
// ActivitySubstate is the persisted activity reading: the state, when it was
// last observed, and which source reported it.
type ActivitySubstate struct {
// Activity captures the persisted activity reading: the state and when it was
// last observed.
type Activity struct {
State ActivityState `json:"state"`
LastActivityAt time.Time `json:"lastActivityAt"`
Source ActivitySource `json:"source"`
}

2
backend/internal/domain/session.go
View File

@ -41,7 +41,7 @@ type SessionRecord struct {
IssueID IssueID `json:"issueId,omitempty"`
Kind SessionKind `json:"kind"`
Harness AgentHarness `json:"harness,omitempty"`
Activity ActivitySubstate `json:"activity"`
Activity Activity `json:"activity"`
IsTerminated bool `json:"isTerminated"`
Metadata SessionMetadata `json:"-"`
CreatedAt time.Time `json:"createdAt"`

6
backend/internal/domain/status.go
View File

@ -16,7 +16,6 @@ const (
StatusMergeable SessionStatus = "mergeable"
StatusMerged SessionStatus = "merged"
StatusNeedsInput SessionStatus = "needs_input"
StatusStuck SessionStatus = "stuck"
StatusIdle SessionStatus = "idle"
StatusTerminated SessionStatus = "terminated"
)
@ -32,11 +31,8 @@ func DeriveStatus(rec SessionRecord, pr *PRFacts) SessionStatus {
return StatusTerminated
}
switch rec.Activity.State {
case ActivityWaitingInput:
if rec.Activity.State == ActivityWaitingInput {
return StatusNeedsInput
case ActivityBlocked:
return StatusStuck
}
if pr != nil {

3
backend/internal/domain/status_test.go
View File

@ -3,7 +3,7 @@ package domain
import "testing"
func rec(activity ActivityState, terminated bool) SessionRecord {
return SessionRecord{Activity: ActivitySubstate{State: activity}, IsTerminated: terminated}
return SessionRecord{Activity: Activity{State: activity}, IsTerminated: terminated}
}
func pr(facts PRFacts) *PRFacts { return &facts }
@ -18,7 +18,6 @@ func TestDeriveStatusFromSessionFactsAndPR(t *testing.T) {
{"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},
{"blocked", rec(ActivityBlocked, false), pr(PRFacts{CI: CIFailing}), StatusStuck},
{"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},

15
backend/internal/lifecycle/manager.go
View File

@ -65,7 +65,7 @@ func (m *Manager) ApplyRuntimeObservation(ctx context.Context, id domain.Session
}
next := cur
next.IsTerminated = true
next.Activity = domain.ActivitySubstate{State: domain.ActivityExited, LastActivityAt: timeOr(f.ObservedAt, now), Source: domain.SourceRuntime}
next.Activity = domain.Activity{State: domain.ActivityExited, LastActivityAt: timeOr(f.ObservedAt, now)}
return next, true
})
}
@ -79,11 +79,8 @@ func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID,
if cur.IsTerminated {
return cur, false
}
if !s.Source.CanOverride(cur.Activity.Source) {
return cur, false
}
next := cur
act := domain.ActivitySubstate{State: s.State, LastActivityAt: timeOr(s.Timestamp, now), Source: s.Source}
act := domain.Activity{State: s.State, LastActivityAt: timeOr(s.Timestamp, now)}
if sameActivity(cur.Activity, act) {
return cur, false
}
@ -108,7 +105,7 @@ func (m *Manager) MarkSpawned(ctx context.Context, id domain.SessionID, metadata
}
now := m.clock()
rec.IsTerminated = false
rec.Activity = domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: now, Source: domain.SourceRuntime}
rec.Activity = domain.Activity{State: domain.ActivityIdle, LastActivityAt: now}
rec.Metadata = mergeMetadata(rec.Metadata, metadata)
rec.UpdatedAt = now
return m.store.UpdateSession(ctx, rec)
@ -121,13 +118,13 @@ func (m *Manager) MarkTerminated(ctx context.Context, id domain.SessionID) error
return cur, false
}
cur.IsTerminated = true
cur.Activity = domain.ActivitySubstate{State: domain.ActivityExited, LastActivityAt: now, Source: domain.SourceRuntime}
cur.Activity = domain.Activity{State: domain.ActivityExited, LastActivityAt: now}
return cur, true
})
}
func sameActivity(a, b domain.ActivitySubstate) bool {
return a.State == b.State && a.Source == b.Source && a.LastActivityAt.Equal(b.LastActivityAt)
func sameActivity(a, b domain.Activity) bool {
return a.State == b.State && a.LastActivityAt.Equal(b.LastActivityAt)
}
func mergeMetadata(base, in domain.SessionMetadata) domain.SessionMetadata {

26
backend/internal/lifecycle/manager_test.go
View File

@ -51,7 +51,7 @@ func newManager() (*Manager, *fakeStore, *fakeMessenger) {
}
func working(id domain.SessionID) domain.SessionRecord {
return domain.SessionRecord{ID: id, ProjectID: "mer", Activity: domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: time.Now(), Source: domain.SourceNative}}
return domain.SessionRecord{ID: id, ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: time.Now()}}
}
func TestRuntimeObservation_InferredDeathSetsTerminated(t *testing.T) {
@ -92,30 +92,6 @@ func TestActivity_InvalidIsIgnored(t *testing.T) {
}
}
func TestActivity_WeakerSourceDoesNotOverrideStronger(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")
before := st.sessions["mer-1"]
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityIdle, Source: domain.SourceRuntime}); err != nil {
t.Fatal(err)
}
if st.sessions["mer-1"] != before {
t.Fatalf("weaker runtime signal should not override native activity, got %+v", st.sessions["mer-1"])
}
}
func TestActivity_StrongerSourceOverridesWeaker(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: time.Now(), Source: domain.SourceRuntime}}
if err := m.ApplyActivitySignal(ctx, "mer-1", ports.ActivitySignal{Valid: true, State: domain.ActivityActive, Source: domain.SourceNative}); err != nil {
t.Fatal(err)
}
got := st.sessions["mer-1"].Activity
if got.State != domain.ActivityActive || got.Source != domain.SourceNative {
t.Fatalf("stronger native signal should override runtime, got %+v", got)
}
}
func TestMarkTerminated(t *testing.T) {
m, st, _ := newManager()
st.sessions["mer-1"] = working("mer-1")

2
backend/internal/lifecycle/reactions.go
View File

@ -39,7 +39,7 @@ func (m *Manager) ApplyPRObservation(ctx context.Context, id domain.SessionID, o
if err != nil || !ok {
return err
}
if rec.IsTerminated || rec.Activity.State == domain.ActivityBlocked || rec.Activity.State == domain.ActivityWaitingInput {
if rec.IsTerminated || rec.Activity.State == domain.ActivityWaitingInput {
return nil
}
if o.CI == domain.CIFailing {

4
backend/internal/lifecycle/runtime.go
View File

@ -9,7 +9,7 @@ import (
const defaultRecentActivityWindow = 60 * time.Second
func hasRecentActivity(a domain.ActivitySubstate, now time.Time, window time.Duration) bool {
func hasRecentActivity(a domain.Activity, now time.Time, window time.Duration) bool {
switch {
case a.State == domain.ActivityExited:
return false
@ -22,7 +22,7 @@ func hasRecentActivity(a domain.ActivitySubstate, now time.Time, window time.Dur
}
}
func runtimeClearlyDead(f ports.RuntimeFacts, activity domain.ActivitySubstate, now time.Time, window time.Duration) bool {
func runtimeClearlyDead(f ports.RuntimeFacts, activity domain.Activity, now time.Time, window time.Duration) bool {
observedAt := timeOr(f.ObservedAt, now)
return f.Probe == ports.ProbeDead && !hasRecentActivity(activity, observedAt, window)
}

2
backend/internal/observe/reaper/reaper_test.go
View File

@ -43,7 +43,7 @@ func (r fakeRuntime) IsAlive(context.Context, ports.RuntimeHandle) (bool, error)
func probableSession(id domain.SessionID) domain.SessionRecord {
return domain.SessionRecord{
ID: id,
Activity: domain.ActivitySubstate{State: domain.ActivityActive},
Activity: domain.Activity{State: domain.ActivityActive},
Metadata: domain.SessionMetadata{RuntimeHandleID: "h1"},
}
}

1
backend/internal/ports/runtime_observations.go
View File

@ -30,5 +30,4 @@ type ActivitySignal struct {
Valid bool
State domain.ActivityState
Timestamp time.Time
Source domain.ActivitySource
}

2
backend/internal/session/manager.go
View File

@ -288,7 +288,7 @@ func seedRecord(cfg ports.SpawnConfig, now time.Time) domain.SessionRecord {
CreatedAt: now,
UpdatedAt: now,
Harness: cfg.Harness,
Activity: domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: now, Source: domain.SourceNone},
Activity: domain.Activity{State: domain.ActivityIdle, LastActivityAt: now},
}
}

10
backend/internal/session/manager_test.go
View File

@ -68,7 +68,7 @@ func (l *fakeLCM) MarkSpawned(_ context.Context, id domain.SessionID, metadata d
l.completed++
rec := l.store.sessions[id]
rec.IsTerminated = false
rec.Activity = domain.ActivitySubstate{State: domain.ActivityIdle, LastActivityAt: time.Now(), Source: domain.SourceRuntime}
rec.Activity = domain.Activity{State: domain.ActivityIdle, LastActivityAt: time.Now()}
rec.Metadata = metadata
l.store.sessions[id] = rec
return nil
@ -76,7 +76,7 @@ func (l *fakeLCM) MarkSpawned(_ context.Context, id domain.SessionID, metadata d
func (l *fakeLCM) MarkTerminated(_ context.Context, id domain.SessionID) error {
rec := l.store.sessions[id]
rec.IsTerminated = true
rec.Activity = domain.ActivitySubstate{State: domain.ActivityExited, LastActivityAt: time.Now(), Source: domain.SourceRuntime}
rec.Activity = domain.Activity{State: domain.ActivityExited, LastActivityAt: time.Now()}
l.store.sessions[id] = rec
return nil
}
@ -134,10 +134,10 @@ func newManager() (*Manager, *fakeStore, *fakeRuntime, *fakeWorkspace) {
return m, st, rt, ws
}
func seedTerminal(st *fakeStore, id domain.SessionID, meta domain.SessionMetadata) {
st.sessions[id] = domain.SessionRecord{ID: id, ProjectID: "mer", Metadata: meta, IsTerminated: true, Activity: domain.ActivitySubstate{State: domain.ActivityExited}}
st.sessions[id] = domain.SessionRecord{ID: id, ProjectID: "mer", Metadata: meta, IsTerminated: true, Activity: domain.Activity{State: domain.ActivityExited}}
}
func mkLive(id domain.SessionID) domain.SessionRecord {
return domain.SessionRecord{ID: id, ProjectID: "mer", Metadata: domain.SessionMetadata{WorkspacePath: "/ws/" + string(id), RuntimeHandleID: "h1"}, Activity: domain.ActivitySubstate{State: domain.ActivityActive}}
return domain.SessionRecord{ID: id, ProjectID: "mer", Metadata: domain.SessionMetadata{WorkspacePath: "/ws/" + string(id), RuntimeHandleID: "h1"}, Activity: domain.Activity{State: domain.ActivityActive}}
}
func TestSpawn_AssignsIDAndGoesIdle(t *testing.T) {
@ -185,7 +185,7 @@ func TestKill_TearsDownRuntimeAndWorkspace(t *testing.T) {
}
func TestKill_RefusesIncompleteHandle(t *testing.T) {
m, st, _, _ := newManager()
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.ActivitySubstate{State: domain.ActivityActive}}
st.sessions["mer-1"] = domain.SessionRecord{ID: "mer-1", ProjectID: "mer", Activity: domain.Activity{State: domain.ActivityActive}}
if _, err := m.Kill(ctx, "mer-1"); !errors.Is(err, ErrIncompleteHandle) {
t.Fatalf("want ErrIncompleteHandle, got %v", err)
}

1
backend/internal/storage/sqlite/gen/models.go
View File

@ -71,7 +71,6 @@ type Session struct {
Harness domain.AgentHarness
ActivityState domain.ActivityState
ActivityLastAt time.Time
ActivitySource domain.ActivitySource
IsTerminated bool
Branch string
WorkspacePath string

19
backend/internal/storage/sqlite/gen/sessions.sql.go
View File

@ -14,7 +14,7 @@ import (
const getSession = `-- name: GetSession :one
SELECT id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated, branch, workspace_path,
activity_state, activity_last_at, is_terminated, branch, workspace_path,
runtime_handle_id, agent_session_id, prompt, created_at, updated_at
FROM sessions WHERE id = ?
`
@ -31,7 +31,6 @@ func (q *Queries) GetSession(ctx context.Context, id domain.SessionID) (Session,
&i.Harness,
&i.ActivityState,
&i.ActivityLastAt,
&i.ActivitySource,
&i.IsTerminated,
&i.Branch,
&i.WorkspacePath,
@ -47,10 +46,10 @@ func (q *Queries) GetSession(ctx context.Context, id domain.SessionID) (Session,
const insertSession = `-- name: InsertSession :exec
INSERT INTO sessions (
id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated,
activity_state, activity_last_at, is_terminated,
branch, workspace_path, runtime_handle_id, agent_session_id, prompt,
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`
type InsertSessionParams struct {
@ -62,7 +61,6 @@ type InsertSessionParams struct {
Harness domain.AgentHarness
ActivityState domain.ActivityState
ActivityLastAt time.Time
ActivitySource domain.ActivitySource
IsTerminated bool
Branch string
WorkspacePath string
@ -83,7 +81,6 @@ func (q *Queries) InsertSession(ctx context.Context, arg InsertSessionParams) er
arg.Harness,
arg.ActivityState,
arg.ActivityLastAt,
arg.ActivitySource,
arg.IsTerminated,
arg.Branch,
arg.WorkspacePath,
@ -98,7 +95,7 @@ func (q *Queries) InsertSession(ctx context.Context, arg InsertSessionParams) er
const listAllSessions = `-- name: ListAllSessions :many
SELECT id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated, branch, workspace_path,
activity_state, activity_last_at, is_terminated, branch, workspace_path,
runtime_handle_id, agent_session_id, prompt, created_at, updated_at
FROM sessions ORDER BY project_id, num
`
@ -121,7 +118,6 @@ func (q *Queries) ListAllSessions(ctx context.Context) ([]Session, error) {
&i.Harness,
&i.ActivityState,
&i.ActivityLastAt,
&i.ActivitySource,
&i.IsTerminated,
&i.Branch,
&i.WorkspacePath,
@ -146,7 +142,7 @@ func (q *Queries) ListAllSessions(ctx context.Context) ([]Session, error) {
const listSessionsByProject = `-- name: ListSessionsByProject :many
SELECT id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated, branch, workspace_path,
activity_state, activity_last_at, is_terminated, branch, workspace_path,
runtime_handle_id, agent_session_id, prompt, created_at, updated_at
FROM sessions WHERE project_id = ? ORDER BY num
`
@ -169,7 +165,6 @@ func (q *Queries) ListSessionsByProject(ctx context.Context, projectID domain.Pr
&i.Harness,
&i.ActivityState,
&i.ActivityLastAt,
&i.ActivitySource,
&i.IsTerminated,
&i.Branch,
&i.WorkspacePath,
@ -206,7 +201,7 @@ func (q *Queries) NextSessionNum(ctx context.Context, projectID domain.ProjectID
const updateSession = `-- name: UpdateSession :exec
UPDATE sessions SET
issue_id = ?, kind = ?, harness = ?,
activity_state = ?, activity_last_at = ?, activity_source = ?, is_terminated = ?,
activity_state = ?, activity_last_at = ?, is_terminated = ?,
branch = ?, workspace_path = ?, runtime_handle_id = ?, agent_session_id = ?, prompt = ?,
updated_at = ?
WHERE id = ?
@ -218,7 +213,6 @@ type UpdateSessionParams struct {
Harness domain.AgentHarness
ActivityState domain.ActivityState
ActivityLastAt time.Time
ActivitySource domain.ActivitySource
IsTerminated bool
Branch string
WorkspacePath string
@ -236,7 +230,6 @@ func (q *Queries) UpdateSession(ctx context.Context, arg UpdateSessionParams) er
arg.Harness,
arg.ActivityState,
arg.ActivityLastAt,
arg.ActivitySource,
arg.IsTerminated,
arg.Branch,
arg.WorkspacePath,

11
backend/internal/storage/sqlite/migrations/0002_remove_activity_source.sql Normal file
View File

@ -0,0 +1,11 @@
-- +goose Up
-- +goose StatementBegin
UPDATE sessions SET activity_state = 'waiting_input' WHERE activity_state = 'blocked';
ALTER TABLE sessions DROP COLUMN activity_source;
-- +goose StatementEnd
-- +goose Down
-- +goose StatementBegin
ALTER TABLE sessions ADD COLUMN activity_source TEXT NOT NULL DEFAULT 'none'
CHECK (activity_source IN ('native', 'terminal', 'hook', 'runtime', 'none'));
-- +goose StatementEnd

12
backend/internal/storage/sqlite/queries/sessions.sql
View File

@ -4,34 +4,34 @@ SELECT COALESCE(MAX(num), 0) + 1 AS next FROM sessions WHERE project_id = ?;
-- name: InsertSession :exec
INSERT INTO sessions (
id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated,
activity_state, activity_last_at, is_terminated,
branch, workspace_path, runtime_handle_id, agent_session_id, prompt,
created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?);
-- name: UpdateSession :exec
UPDATE sessions SET
issue_id = ?, kind = ?, harness = ?,
activity_state = ?, activity_last_at = ?, activity_source = ?, is_terminated = ?,
activity_state = ?, activity_last_at = ?, is_terminated = ?,
branch = ?, workspace_path = ?, runtime_handle_id = ?, agent_session_id = ?, prompt = ?,
updated_at = ?
WHERE id = ?;
-- name: GetSession :one
SELECT id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated, branch, workspace_path,
activity_state, activity_last_at, is_terminated, branch, workspace_path,
runtime_handle_id, agent_session_id, prompt, created_at, updated_at
FROM sessions WHERE id = ?;
-- name: ListSessionsByProject :many
SELECT id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated, branch, workspace_path,
activity_state, activity_last_at, is_terminated, branch, workspace_path,
runtime_handle_id, agent_session_id, prompt, created_at, updated_at
FROM sessions WHERE project_id = ? ORDER BY num;
-- name: ListAllSessions :many
SELECT id, project_id, num, issue_id, kind, harness,
activity_state, activity_last_at, activity_source, is_terminated, branch, workspace_path,
activity_state, activity_last_at, is_terminated, branch, workspace_path,
runtime_handle_id, agent_session_id, prompt, created_at, updated_at
FROM sessions ORDER BY project_id, num;

10
backend/internal/storage/sqlite/store/session_store.go
View File

@ -85,10 +85,9 @@ func rowToRecord(row gen.Session) domain.SessionRecord {
IssueID: row.IssueID,
Kind: row.Kind,
Harness: row.Harness,
Activity: domain.ActivitySubstate{
Activity: domain.Activity{
State: row.ActivityState,
LastActivityAt: row.ActivityLastAt,
Source: row.ActivitySource,
},
IsTerminated: row.IsTerminated,
Metadata: domain.SessionMetadata{
@ -114,7 +113,6 @@ func recordToInsert(rec domain.SessionRecord, num int64) gen.InsertSessionParams
Harness: rec.Harness,
ActivityState: activity.State,
ActivityLastAt: activity.LastActivityAt,
ActivitySource: activity.Source,
IsTerminated: rec.IsTerminated,
Branch: rec.Metadata.Branch,
WorkspacePath: rec.Metadata.WorkspacePath,
@ -135,7 +133,6 @@ func recordToUpdate(rec domain.SessionRecord) gen.UpdateSessionParams {
Harness: rec.Harness,
ActivityState: activity.State,
ActivityLastAt: activity.LastActivityAt,
ActivitySource: activity.Source,
IsTerminated: rec.IsTerminated,
Branch: rec.Metadata.Branch,
WorkspacePath: rec.Metadata.WorkspacePath,
@ -146,13 +143,10 @@ func recordToUpdate(rec domain.SessionRecord) gen.UpdateSessionParams {
}
}
func normalActivity(a domain.ActivitySubstate, fallback time.Time) domain.ActivitySubstate {
func normalActivity(a domain.Activity, fallback time.Time) domain.Activity {
if a.State == "" {
a.State = domain.ActivityIdle
}
if a.Source == "" {
a.Source = domain.SourceNone
}
if a.LastActivityAt.IsZero() {
a.LastActivityAt = fallback
}

4
backend/internal/storage/sqlite/store/store_test.go
View File

@ -37,7 +37,7 @@ func sampleRecord(project string) domain.SessionRecord {
ProjectID: domain.ProjectID(project),
Kind: domain.KindWorker,
Harness: domain.HarnessClaudeCode,
Activity: domain.ActivitySubstate{State: domain.ActivityActive, LastActivityAt: now, Source: domain.SourceNative},
Activity: domain.Activity{State: domain.ActivityActive, LastActivityAt: now},
Metadata: domain.SessionMetadata{Branch: "feat/x", WorkspacePath: "/ws"},
CreatedAt: now,
UpdatedAt: now,
@ -108,7 +108,7 @@ func TestSessionUpdateActivityAndTermination(t *testing.T) {
seedProject(t, s, "mer")
r, _ := s.CreateSession(ctx, sampleRecord("mer"))
r.Activity = domain.ActivitySubstate{State: domain.ActivityWaitingInput, LastActivityAt: r.CreatedAt, Source: domain.SourceHook}
r.Activity = domain.Activity{State: domain.ActivityWaitingInput, LastActivityAt: r.CreatedAt}
r.IsTerminated = true
if err := s.UpdateSession(ctx, r); err != nil {
t.Fatal(err)

4
backend/sqlc.yaml
View File

@ -84,7 +84,3 @@ sql:
go_type:
import: "github.com/aoagents/agent-orchestrator/backend/internal/domain"
type: "ActivityState"
- column: "sessions.activity_source"
go_type:
import: "github.com/aoagents/agent-orchestrator/backend/internal/domain"
type: "ActivitySource"

10
docs/architecture.md
View File

@ -13,8 +13,7 @@ OBSERVE external facts → UPDATE durable facts → DERIVE display status / ACT
The durable session facts are:
- `activity_state` — what the agent last reported or what the runtime observer
can safely conclude (`active`, `ready`, `idle`, `waiting_input`, `blocked`,
`exited`).
can safely conclude (`active`, `idle`, `waiting_input`, `exited`).
- `is_terminated` — whether the session should be treated as over.
- PR facts in the `pr`, `pr_checks`, and `pr_comment` tables.
@ -43,11 +42,10 @@ backend/internal/adapters Zellij/git-worktree/GitHub adapters
1. `is_terminated` → `terminated`, except merged PRs display `merged`.
2. `activity_state=waiting_input` → `needs_input`.
3. `activity_state=blocked` → `stuck`.
4. Open PR facts drive PR pipeline statuses: `ci_failed`, `draft`,
3. Open PR facts drive PR pipeline statuses: `ci_failed`, `draft`,
`changes_requested`, `mergeable`, `approved`, `review_pending`, `pr_open`.
5. `activity_state=active` → `working`.
6. Everything else → `idle`.
4. `activity_state=active` → `working`.
5. Everything else → `idle`.
## Lifecycle manager