diff --git a/backend/internal/terminal/manager.go b/backend/internal/terminal/manager.go index 068f36bf2..986d9d921 100644 --- a/backend/internal/terminal/manager.go +++ b/backend/internal/terminal/manager.go @@ -284,9 +284,9 @@ func (c *connState) handleSubscribe() { Type: msgSnapshot, Session: &sessionUpdate{ Seq: e.Seq, + ProjectID: e.ProjectID, SessionID: e.SessionID, - EventType: e.EventType, - Revision: e.Revision, + EventType: string(e.Type), }, }) }) diff --git a/backend/internal/terminal/manager_test.go b/backend/internal/terminal/manager_test.go index 6169c34fd..71187ed7b 100644 --- a/backend/internal/terminal/manager_test.go +++ b/backend/internal/terminal/manager_test.go @@ -127,7 +127,7 @@ func TestServeForwardsSessionChannelFromCDC(t *testing.T) { conn.in <- clientMsg{Ch: chSubscribe, Type: msgSubscribe} // Give the subscription time to register before publishing. eventually(t, time.Second, func() bool { - bc.Publish(cdc.Event{Seq: 9, SessionID: "s1", EventType: "session_updated", Revision: 4}) + bc.Publish(cdc.Event{Seq: 9, ProjectID: "p1", SessionID: "s1", Type: cdc.EventSessionUpdated}) select { case m := <-conn.out: return m.Ch == chSessions && m.Session != nil && m.Session.Seq == 9 diff --git a/backend/internal/terminal/protocol.go b/backend/internal/terminal/protocol.go index 472e03872..31a47999a 100644 --- a/backend/internal/terminal/protocol.go +++ b/backend/internal/terminal/protocol.go @@ -65,7 +65,7 @@ type serverMsg struct { // change_log payload blob; the client refetches detail over the REST surface. type sessionUpdate struct { Seq int64 `json:"seq"` - SessionID string `json:"sessionId"` + ProjectID string `json:"projectId"` + SessionID string `json:"sessionId,omitempty"` EventType string `json:"eventType"` - Revision int64 `json:"revision"` } diff --git a/backend/internal/terminal/protocol_test.go b/backend/internal/terminal/protocol_test.go index ef521ed8a..e42170d5d 100644 --- a/backend/internal/terminal/protocol_test.go +++ b/backend/internal/terminal/protocol_test.go @@ -33,7 +33,7 @@ func TestServerMsgSessionFrameWireShape(t *testing.T) { Ch: chSessions, Type: msgSnapshot, Session: &sessionUpdate{ - Seq: 7, SessionID: "s1", EventType: "session_updated", Revision: 3, + Seq: 7, ProjectID: "p1", SessionID: "s1", EventType: "session_updated", }, } raw, err := json.Marshal(msg) @@ -41,7 +41,7 @@ func TestServerMsgSessionFrameWireShape(t *testing.T) { t.Fatalf("marshal: %v", err) } // Golden wire shape the client depends on. - want := `{"ch":"sessions","type":"snapshot","session":{"seq":7,"sessionId":"s1","eventType":"session_updated","revision":3}}` + want := `{"ch":"sessions","type":"snapshot","session":{"seq":7,"projectId":"p1","sessionId":"s1","eventType":"session_updated"}}` if string(raw) != want { t.Fatalf("wire shape:\n got %s\nwant %s", raw, want) } diff --git a/backend/main.go b/backend/main.go index 7840285fc..950ea4e35 100644 --- a/backend/main.go +++ b/backend/main.go @@ -12,10 +12,12 @@ import ( "os/signal" "syscall" + "github.com/aoagents/agent-orchestrator/backend/internal/adapters/runtime/tmux" "github.com/aoagents/agent-orchestrator/backend/internal/config" "github.com/aoagents/agent-orchestrator/backend/internal/httpd" "github.com/aoagents/agent-orchestrator/backend/internal/runfile" "github.com/aoagents/agent-orchestrator/backend/internal/storage/sqlite" + "github.com/aoagents/agent-orchestrator/backend/internal/terminal" ) func main() { @@ -42,11 +44,6 @@ func run() error { return fmt.Errorf("daemon already running (pid %d, port %d); refusing to start", live.PID, live.Port) } - srv, err := httpd.New(cfg, log, nil) - if err != nil { - return err - } - // Open the durable store and bring up the CDC substrate: the DB triggers // capture changes into change_log, the poller tails it, and the broadcaster // fans events out to the SSE transport. The LCM/Session Manager and the HTTP @@ -70,6 +67,19 @@ func run() error { return err } + // Terminal streaming: the tmux runtime supplies the PTY-attach command and + // liveness; the CDC broadcaster feeds the session-state channel. The manager + // is handed to httpd, which mounts it at /mux. Raw PTY bytes never flow + // through the CDC change_log — only session-state events do. + runtimeAdapter := tmux.New(tmux.Options{}) + termMgr := terminal.NewManager(runtimeAdapter, cdcPipe.Broadcaster, log) + defer termMgr.Close() + + srv, err := httpd.New(cfg, log, termMgr) + if err != nil { + return err + } + // Bring up the Lifecycle Manager (sole store writer) and the reaper (OBSERVE // timer). This makes the write path live end-to-end: LCM write -> store -> DB // trigger -> change_log -> poller -> broadcaster. The collaborators it needs