Skip to content

Commit 046fd4f

Browse files
nathanwhitnathanwhit
committed
Add session tail data to the live session view
1 parent f517b76 commit 046fd4f

9 files changed

Lines changed: 650 additions & 102 deletions

File tree

‎internal/core/types.go‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -472,6 +472,20 @@ type SessionCurrentAction struct {
472472
EventID int64 `json:"eventId,omitempty"`
473473
}
474474

475+
type SessionChangedFile struct {
476+
Path string `json:"path"`
477+
Status string `json:"status,omitempty"`
478+
}
479+
480+
type SessionCompletion struct {
481+
Status WorkerStatus `json:"status,omitempty"`
482+
Summary string `json:"summary,omitempty"`
483+
Error string `json:"error,omitempty"`
484+
EventID int64 `json:"eventId,omitempty"`
485+
At time.Time `json:"at,omitempty"`
486+
ChangedFiles []SessionChangedFile `json:"changedFiles,omitempty"`
487+
}
488+
475489
type SessionTail struct {
476490
SessionID string `json:"sessionId"`
477491
WorkerID string `json:"workerId"`
@@ -480,6 +494,12 @@ type SessionTail struct {
480494
LastEventID int64 `json:"lastEventId"`
481495
Events []Event `json:"events"`
482496
CurrentAction *SessionCurrentAction `json:"currentAction,omitempty"`
497+
Session *Session `json:"session,omitempty"`
498+
Worker *Worker `json:"worker,omitempty"`
499+
Node *ExecutionNode `json:"node,omitempty"`
500+
PullRequests []PullRequest `json:"pullRequests,omitempty"`
501+
Completion *SessionCompletion `json:"completion,omitempty"`
502+
ChangedFiles []SessionChangedFile `json:"changedFiles,omitempty"`
483503
}
484504

485505
type TargetCapacity struct {

‎internal/httpapi/server_test.go‎

Lines changed: 52 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -587,6 +587,18 @@ func TestSessionTailEndpointReturnsWorkerScopedEvents(t *testing.T) {
587587
if _, err := store.Append(ctx, core.Event{Type: core.EventWorkerCreated, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{"kind": "mock"})}); err != nil {
588588
t.Fatal(err)
589589
}
590+
if _, err := store.Append(ctx, core.Event{Type: core.EventExecutionPlanned, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{
591+
"nodeId": "node-session-tail",
592+
"workerId": workerID,
593+
"workerKind": "mock",
594+
"role": "implementation",
595+
"targetId": "vultr-vm",
596+
"targetKind": "ssh",
597+
"remoteSession": "aged-worker-tail",
598+
"remoteWorkDir": "/work/repo",
599+
})}); err != nil {
600+
t.Fatal(err)
601+
}
590602
started, err := store.Append(ctx, core.Event{Type: core.EventWorkerStarted, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{})})
591603
if err != nil {
592604
t.Fatal(err)
@@ -598,6 +610,27 @@ func TestSessionTailEndpointReturnsWorkerScopedEvents(t *testing.T) {
598610
if err != nil {
599611
t.Fatal(err)
600612
}
613+
completed, err := store.Append(ctx, core.Event{Type: core.EventWorkerCompleted, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{
614+
"status": core.WorkerSucceeded,
615+
"summary": "session done",
616+
"changedFiles": []map[string]any{{"path": "web/src/main.tsx", "status": "modified"}},
617+
})})
618+
if err != nil {
619+
t.Fatal(err)
620+
}
621+
if _, err := store.Append(ctx, core.Event{Type: core.EventPRPublished, TaskID: taskID, Payload: core.MustJSON(map[string]any{
622+
"id": "pr-session-tail",
623+
"repo": "owner/repo",
624+
"number": 9,
625+
"url": "https://github.com/owner/repo/pull/9",
626+
"branch": "session-tail",
627+
"title": "Session tail",
628+
"metadata": map[string]any{
629+
"workerId": workerID,
630+
},
631+
})}); err != nil {
632+
t.Fatal(err)
633+
}
601634

602635
service := orchestrator.NewService(store, orchestrator.StaticBrain{WorkerKind: "mock"}, worker.DefaultRunners(), t.TempDir())
603636
server := httptest.NewServer(New(service, nil).Routes())
@@ -618,15 +651,30 @@ func TestSessionTailEndpointReturnsWorkerScopedEvents(t *testing.T) {
618651
if tail.SessionID != workerID || tail.WorkerID != workerID || tail.TaskID != taskID {
619652
t.Fatalf("tail identity = %+v", tail)
620653
}
621-
if tail.LastEventID != output.ID {
622-
t.Fatalf("lastEventId = %d, want %d", tail.LastEventID, output.ID)
654+
if tail.LastEventID != completed.ID {
655+
t.Fatalf("lastEventId = %d, want %d", tail.LastEventID, completed.ID)
623656
}
624-
if len(tail.Events) != 1 || tail.Events[0].ID != output.ID {
625-
t.Fatalf("events = %+v, want output %d", tail.Events, output.ID)
657+
if len(tail.Events) != 2 || tail.Events[0].ID != output.ID || tail.Events[1].ID != completed.ID {
658+
t.Fatalf("events = %+v, want output/completed %d/%d", tail.Events, output.ID, completed.ID)
626659
}
627660
if tail.CurrentAction == nil || !strings.Contains(tail.CurrentAction.Text, "session output") {
628661
t.Fatalf("current action = %+v", tail.CurrentAction)
629662
}
663+
if tail.Session == nil || tail.Session.RemoteSession != "aged-worker-tail" {
664+
t.Fatalf("session context = %+v", tail.Session)
665+
}
666+
if tail.Worker == nil || tail.Worker.Kind != "mock" {
667+
t.Fatalf("worker context = %+v", tail.Worker)
668+
}
669+
if tail.Node == nil || tail.Node.ID != "node-session-tail" {
670+
t.Fatalf("node context = %+v", tail.Node)
671+
}
672+
if len(tail.PullRequests) != 1 || tail.PullRequests[0].ID != "pr-session-tail" {
673+
t.Fatalf("pull requests = %+v", tail.PullRequests)
674+
}
675+
if tail.Completion == nil || tail.Completion.EventID != completed.ID || len(tail.ChangedFiles) != 1 || tail.ChangedFiles[0].Path != "web/src/main.tsx" {
676+
t.Fatalf("completion = %+v changedFiles = %+v", tail.Completion, tail.ChangedFiles)
677+
}
630678
}
631679

632680
func TestSessionControlEndpointsDelegateToWorker(t *testing.T) {

‎internal/orchestrator/service.go‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1732,6 +1732,10 @@ func (s *Service) SessionTail(ctx context.Context, sessionID string, afterID int
17321732
if err != nil {
17331733
return core.SessionTail{}, err
17341734
}
1735+
snapshot, err := s.store.Snapshot(ctx)
1736+
if err != nil {
1737+
return core.SessionTail{}, err
1738+
}
17351739
if len(kinds) == 0 {
17361740
kinds = []core.EventType{
17371741
core.EventWorkerOutput,
@@ -1759,6 +1763,14 @@ func (s *Service) SessionTail(ctx context.Context, sessionID string, afterID int
17591763
EventID: session.CurrentActionEvent,
17601764
}
17611765
}
1766+
worker := sessionTailWorker(snapshot.Workers, session.WorkerID)
1767+
node := sessionTailNode(snapshot.ExecutionNodes, session)
1768+
pullRequests := sessionTailPullRequests(snapshot.PullRequests, session.TaskID, session.WorkerID)
1769+
completion := s.sessionTailCompletion(ctx, session.WorkerID)
1770+
changedFiles := []core.SessionChangedFile(nil)
1771+
if completion != nil {
1772+
changedFiles = completion.ChangedFiles
1773+
}
17621774
return core.SessionTail{
17631775
SessionID: session.ID,
17641776
WorkerID: session.WorkerID,
@@ -1767,9 +1779,101 @@ func (s *Service) SessionTail(ctx context.Context, sessionID string, afterID int
17671779
LastEventID: lastEventID,
17681780
Events: events,
17691781
CurrentAction: currentAction,
1782+
Session: &session,
1783+
Worker: worker,
1784+
Node: node,
1785+
PullRequests: pullRequests,
1786+
Completion: completion,
1787+
ChangedFiles: changedFiles,
17701788
}, nil
17711789
}
17721790

1791+
func sessionTailWorker(workers []core.Worker, workerID string) *core.Worker {
1792+
for _, worker := range workers {
1793+
if worker.ID == workerID {
1794+
return &worker
1795+
}
1796+
}
1797+
return nil
1798+
}
1799+
1800+
func sessionTailNode(nodes []core.ExecutionNode, session core.Session) *core.ExecutionNode {
1801+
for _, node := range nodes {
1802+
if session.NodeID != "" && node.ID == session.NodeID {
1803+
return &node
1804+
}
1805+
}
1806+
for _, node := range nodes {
1807+
if node.WorkerID == session.WorkerID {
1808+
return &node
1809+
}
1810+
}
1811+
return nil
1812+
}
1813+
1814+
func sessionTailPullRequests(pullRequests []core.PullRequest, taskID string, workerID string) []core.PullRequest {
1815+
taskPullRequests := []core.PullRequest{}
1816+
workerPullRequests := []core.PullRequest{}
1817+
for _, pr := range pullRequests {
1818+
if pr.TaskID != taskID {
1819+
continue
1820+
}
1821+
taskPullRequests = append(taskPullRequests, pr)
1822+
if workerID != "" && pullRequestMetadataString(pr, "workerId") == workerID {
1823+
workerPullRequests = append(workerPullRequests, pr)
1824+
}
1825+
}
1826+
if len(workerPullRequests) > 0 {
1827+
return workerPullRequests
1828+
}
1829+
return taskPullRequests
1830+
}
1831+
1832+
func (s *Service) sessionTailCompletion(ctx context.Context, workerID string) *core.SessionCompletion {
1833+
events, err := s.store.ListWorkerEvents(ctx, workerID, 0, 1000, core.EventWorkerCompleted)
1834+
if err != nil {
1835+
return nil
1836+
}
1837+
var completion *core.SessionCompletion
1838+
for _, event := range events {
1839+
next := sessionCompletionFromEvent(event)
1840+
if next != nil {
1841+
completion = next
1842+
}
1843+
}
1844+
return completion
1845+
}
1846+
1847+
func sessionCompletionFromEvent(event core.Event) *core.SessionCompletion {
1848+
if event.Type != core.EventWorkerCompleted {
1849+
return nil
1850+
}
1851+
var payload struct {
1852+
Status core.WorkerStatus `json:"status,omitempty"`
1853+
Summary string `json:"summary,omitempty"`
1854+
Error string `json:"error,omitempty"`
1855+
ChangedFiles []core.SessionChangedFile `json:"changedFiles,omitempty"`
1856+
WorkspaceChanges struct {
1857+
ChangedFiles []core.SessionChangedFile `json:"changedFiles,omitempty"`
1858+
} `json:"workspaceChanges,omitempty"`
1859+
}
1860+
if err := json.Unmarshal(event.Payload, &payload); err != nil {
1861+
return nil
1862+
}
1863+
changedFiles := payload.ChangedFiles
1864+
if len(changedFiles) == 0 {
1865+
changedFiles = payload.WorkspaceChanges.ChangedFiles
1866+
}
1867+
return &core.SessionCompletion{
1868+
Status: payload.Status,
1869+
Summary: payload.Summary,
1870+
Error: payload.Error,
1871+
EventID: event.ID,
1872+
At: event.At,
1873+
ChangedFiles: changedFiles,
1874+
}
1875+
}
1876+
17731877
func (s *Service) Subscribe() (int, <-chan core.Event) {
17741878
return s.broker.Subscribe()
17751879
}

‎internal/orchestrator/service_test.go‎

Lines changed: 54 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -14242,6 +14242,18 @@ func TestServiceSessionTailReturnsWorkerEventsAndCurrentAction(t *testing.T) {
1424214242
if _, err := store.Append(ctx, core.Event{Type: core.EventWorkerCreated, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{"kind": "mock"})}); err != nil {
1424314243
t.Fatal(err)
1424414244
}
14245+
if _, err := store.Append(ctx, core.Event{Type: core.EventExecutionPlanned, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{
14246+
"nodeId": "node-session-tail",
14247+
"workerId": workerID,
14248+
"workerKind": "mock",
14249+
"role": "implementation",
14250+
"targetId": "vultr-vm",
14251+
"targetKind": "ssh",
14252+
"remoteSession": "aged-worker-tail",
14253+
"remoteWorkDir": "/work/repo",
14254+
})}); err != nil {
14255+
t.Fatal(err)
14256+
}
1424514257
started, err := store.Append(ctx, core.Event{Type: core.EventWorkerStarted, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{})})
1424614258
if err != nil {
1424714259
t.Fatal(err)
@@ -14253,6 +14265,29 @@ func TestServiceSessionTailReturnsWorkerEventsAndCurrentAction(t *testing.T) {
1425314265
if err != nil {
1425414266
t.Fatal(err)
1425514267
}
14268+
completed, err := store.Append(ctx, core.Event{Type: core.EventWorkerCompleted, TaskID: taskID, WorkerID: workerID, Payload: core.MustJSON(map[string]any{
14269+
"status": core.WorkerSucceeded,
14270+
"summary": "tail done",
14271+
"workspaceChanges": map[string]any{
14272+
"changedFiles": []map[string]any{{"path": "internal/orchestrator/service.go", "status": "modified"}},
14273+
},
14274+
})})
14275+
if err != nil {
14276+
t.Fatal(err)
14277+
}
14278+
if _, err := store.Append(ctx, core.Event{Type: core.EventPRPublished, TaskID: taskID, Payload: core.MustJSON(map[string]any{
14279+
"id": "pr-session-tail",
14280+
"repo": "owner/repo",
14281+
"number": 9,
14282+
"url": "https://github.com/owner/repo/pull/9",
14283+
"branch": "session-tail",
14284+
"title": "Session tail",
14285+
"metadata": map[string]any{
14286+
"workerId": workerID,
14287+
},
14288+
})}); err != nil {
14289+
t.Fatal(err)
14290+
}
1425614291

1425714292
tail, err := service.SessionTail(ctx, workerID, started.ID, 10)
1425814293
if err != nil {
@@ -14261,15 +14296,30 @@ func TestServiceSessionTailReturnsWorkerEventsAndCurrentAction(t *testing.T) {
1426114296
if tail.SessionID != workerID || tail.WorkerID != workerID || tail.TaskID != taskID {
1426214297
t.Fatalf("tail identity = %+v", tail)
1426314298
}
14264-
if tail.LastEventID != output.ID {
14265-
t.Fatalf("lastEventId = %d, want %d", tail.LastEventID, output.ID)
14299+
if tail.LastEventID != completed.ID {
14300+
t.Fatalf("lastEventId = %d, want %d", tail.LastEventID, completed.ID)
1426614301
}
14267-
if len(tail.Events) != 1 || tail.Events[0].ID != output.ID {
14268-
t.Fatalf("events = %+v, want output event %d", tail.Events, output.ID)
14302+
if len(tail.Events) != 2 || tail.Events[0].ID != output.ID || tail.Events[1].ID != completed.ID {
14303+
t.Fatalf("events = %+v, want output/completed events %d/%d", tail.Events, output.ID, completed.ID)
1426914304
}
1427014305
if tail.CurrentAction == nil || !strings.Contains(tail.CurrentAction.Text, "go test") || tail.CurrentAction.EventID != output.ID {
1427114306
t.Fatalf("current action = %+v", tail.CurrentAction)
1427214307
}
14308+
if tail.Session == nil || tail.Session.RemoteSession != "aged-worker-tail" || tail.Session.RemoteWorkDir != "/work/repo" {
14309+
t.Fatalf("session context = %+v", tail.Session)
14310+
}
14311+
if tail.Worker == nil || tail.Worker.Kind != "mock" {
14312+
t.Fatalf("worker context = %+v", tail.Worker)
14313+
}
14314+
if tail.Node == nil || tail.Node.ID != "node-session-tail" || tail.Node.TargetID != "vultr-vm" {
14315+
t.Fatalf("node context = %+v", tail.Node)
14316+
}
14317+
if len(tail.PullRequests) != 1 || tail.PullRequests[0].ID != "pr-session-tail" {
14318+
t.Fatalf("pull requests = %+v", tail.PullRequests)
14319+
}
14320+
if tail.Completion == nil || tail.Completion.EventID != completed.ID || len(tail.ChangedFiles) != 1 || tail.ChangedFiles[0].Path != "internal/orchestrator/service.go" {
14321+
t.Fatalf("completion = %+v changedFiles = %+v", tail.Completion, tail.ChangedFiles)
14322+
}
1427314323
}
1427414324

1427514325
func TestServiceSessionControlDelegatesToWorker(t *testing.T) {

‎web/e2e/manager-summary.spec.ts‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,19 @@ async function mockApi(page: Page) {
141141
await page.route(`**/api/tasks/${taskId}`, (route) => route.fulfill({ json: snapshot }));
142142
await page.route(`**/api/tasks/${taskId}/assignments`, (route) => route.fulfill({ json: { taskId, assignments: [] } }));
143143
await page.route(`**/api/tasks/${taskId}/events?**`, (route) => route.fulfill({ json: [] }));
144+
await page.route("**/api/sessions/session-manager-summary/tail?**", (route) =>
145+
route.fulfill({
146+
json: {
147+
sessionId: "session-manager-summary",
148+
workerId: "worker-manager-summary",
149+
taskId,
150+
status: "running",
151+
lastEventId: 1,
152+
events: [],
153+
currentAction: { label: "Validation", text: "Running validation checks for manager summary", at: now, eventId: 1 },
154+
},
155+
}),
156+
);
144157
await page.route("**/api/events/stream?**", (route) =>
145158
route.fulfill({
146159
status: 200,
@@ -180,3 +193,18 @@ test("manager summary badges render through dashboard and task detail without ho
180193
await expect(page.getByText("Assignments")).toBeVisible();
181194
await expectNoHorizontalOverflow(page);
182195
});
196+
197+
test("selected task defaults to manager console without legacy backend panes", async ({ page }) => {
198+
await mockApi(page);
199+
await page.setViewportSize({ width: 1366, height: 900 });
200+
await page.goto("/");
201+
202+
await expect(page.getByRole("heading", { name: "Manager summary validation" })).toBeVisible();
203+
await expect(page.getByText("Pull Requests", { exact: true })).toHaveCount(1);
204+
await expect(page.getByRole("heading", { name: "Current State" })).toHaveCount(0);
205+
await expect(page.getByRole("heading", { name: "Orchestration" })).toHaveCount(0);
206+
await expect(page.getByRole("heading", { name: "Worker Detail" })).toHaveCount(0);
207+
await expect(page.getByRole("heading", { name: "Timeline" })).toHaveCount(0);
208+
await expect(page.locator(".debug-backend-internals")).toHaveCount(0);
209+
await expect(page.locator(".debug-pane summary").getByText("Debug")).toBeVisible();
210+
});

0 commit comments

Comments
 (0)