tutti/packages/agent/daemon/activity/functional_test.go

package agentsessionstore_test

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"testing"

	agentsessionstore "github.com/tutti-os/tutti/packages/agent/daemon/activity"
	canonical "github.com/tutti-os/tutti/packages/agent/store-sqlite/canonical"
)

func TestWorkspaceAgentMessageFunctionalFlow(t *testing.T) {
	t.Parallel()

	var acceptedActivity int
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		switch {
		case r.Method == http.MethodPost && r.URL.Path == "/v1/rooms/ws-1/agents/sessions/codex-1/state":
			var body struct {
				State map[string]any `json:"state"`
			}
			if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
				t.Fatalf("decode state body: %v", err)
			}
			if len(body.State) == 0 {
				t.Fatal("state body is empty")
			}
			acceptedActivity++
			_ = json.NewEncoder(w).Encode(map[string]bool{"accepted": true})
		case r.Method == http.MethodPost && r.URL.Path == "/v1/rooms/ws-1/agents/sessions/codex-1/messages":
			var body struct {
				Updates []map[string]any `json:"updates"`
			}
			if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
				t.Fatalf("decode messages body: %v", err)
			}
			acceptedActivity += len(body.Updates)
			_ = json.NewEncoder(w).Encode(map[string]int{"acceptedCount": len(body.Updates)})
		case r.Method == http.MethodGet && r.URL.Path == "/v1/rooms/ws-1/agents/list":
			_, _ = w.Write([]byte(`{
				"presences":[{"id":1,"roomId":"ws-1","userId":"user-1","provider":"codex","status":"working"}],
				"sessions":[{"id":2,"agentSessionId":"codex-1","presenceId":1,"providerSessionId":"codex-1","cwd":"/workspace/ws-1","effectiveStatus":"working","sessionOrigin":"WORKSPACE_AGENT_SESSION_ORIGIN_RUNTIME"}]
			}`))
		case r.Method == http.MethodGet && r.URL.Path == "/v1/rooms/ws-1/agents/sessions/codex-1/messages":
			if r.URL.Query().Get("after_version") != "0" && r.URL.Query().Get("after_version") != "" {
				t.Fatalf("unexpected after_version = %q", r.URL.Query().Get("after_version"))
			}
			if r.URL.Query().Get("session_origin") != agentsessionstore.WorkspaceAgentSessionOriginRuntime {
				t.Fatalf("session_origin = %q", r.URL.Query().Get("session_origin"))
			}
			_, _ = w.Write([]byte(`{
				"messages":[
					{"id":7,"agentSessionId":"agent-session-1","messageId":"msg-1","role":"assistant","kind":"text","payload":{"content":"Done."},"occurredAtUnixMs":1776934803000,"createdAtUnixMs":1776934803100,"version":7},
					{"id":8,"agentSessionId":"agent-session-1","messageId":"act-1","role":"tool","kind":"tool_call","status":"running","payload":{"callId":"call-1","name":"exec_command"},"occurredAtUnixMs":1776934804000,"createdAtUnixMs":1776934804100,"version":8}
				],
				"latestVersion":8
			}`))
		default:
			http.NotFound(w, r)
		}
	}))
	defer server.Close()

	client := agentsessionstore.NewClient(agentsessionstore.Config{BaseURL: server.URL, UserID: "user-1"})

	reply, err := client.ReportActivity(context.Background(), agentsessionstore.ReportActivityInput{
		WorkspaceID: "ws-1",
		StatePatches: []agentsessionstore.WorkspaceAgentStatePatch{{
			AgentSessionID:    "codex-1",
			Provider:          "codex",
			ProviderSessionID: "codex-1",
			CWD:               "/workspace/ws-1",
			CurrentPhase:      "working",
			LifecycleStatus:   "active",
			OccurredAtUnixMS:  1776934800000,
		}},
		MessageUpdates: []agentsessionstore.WorkspaceAgentMessageUpdate{
			{
				AgentSessionID:   "codex-1",
				MessageID:        "msg-user-1",
				TurnID:           "turn-1",
				Role:             "user",
				Kind:             "text",
				Payload:          map[string]any{"content": "Inspect README."},
				OccurredAtUnixMS: 1776934801000,
			},
			{
				AgentSessionID:   "codex-1",
				MessageID:        "msg-assistant-1",
				TurnID:           "turn-1",
				Role:             "assistant",
				Kind:             "text",
				Payload:          map[string]any{"content": "Done."},
				OccurredAtUnixMS: 1776934803000,
			},
		},
		Source: canonical.EventSource{
			SessionOrigin: agentsessionstore.WorkspaceAgentSessionOriginRuntime,
			Provider:      "codex",
		},
	})
	if err != nil {
		t.Fatalf("ReportActivity() error = %v", err)
	}
	acceptedReply := reply.AcceptedStatePatchCount + reply.AcceptedMessageUpdateCount
	if acceptedReply != acceptedActivity || acceptedActivity == 0 {
		t.Fatalf("accepted = reply:%d server:%d want nonzero matching count", acceptedReply, acceptedActivity)
	}

	poller := agentsessionstore.NewPoller(client, agentsessionstore.PollerOptions{Limit: 20})
	result, err := poller.Poll(context.Background(), "ws-1", nil)
	if err != nil {
		t.Fatalf("Poll() error = %v", err)
	}
	if len(result.Snapshot.Presences) != 1 || result.Snapshot.Presences[0].Provider != "codex" {
		t.Fatalf("snapshot presences = %#v", result.Snapshot.Presences)
	}
	messages := result.Messages["codex-1"]
	if len(messages.Messages) != 2 {
		t.Fatalf("messages = %#v", messages.Messages)
	}
	if messages.Messages[0].Kind != "text" ||
		messages.Messages[0].Role != "assistant" ||
		messages.Messages[0].Payload["content"] != "Done." ||
		messages.Messages[0].OccurredAtUnixMS != 1776934803000 ||
		messages.Messages[0].CreatedAtUnixMS != 1776934803100 {
		t.Fatalf("message entry = %#v", messages.Messages[0])
	}
	if messages.Messages[1].Kind != "tool_call" ||
		messages.Messages[1].Role != "tool" ||
		messages.Messages[1].Payload["callId"] != "call-1" ||
		messages.Messages[1].Payload["name"] != "exec_command" ||
		messages.Messages[1].OccurredAtUnixMS != 1776934804000 ||
		messages.Messages[1].CreatedAtUnixMS != 1776934804100 {
		t.Fatalf("tool message entry = %#v", messages.Messages[1])
	}
	if result.Cursors["codex-1"].AfterVersion != 8 {
		t.Fatalf("cursor = %#v", result.Cursors["codex-1"])
	}
}