//go:build e2e

package e2e_test

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api/admin"
	"github.com/tidbcloud/AgentTeam/internal/api/agent"
	"github.com/tidbcloud/AgentTeam/internal/api/app"
	"github.com/tidbcloud/AgentTeam/internal/api/auth"
	"github.com/tidbcloud/AgentTeam/internal/api/channel"
	"github.com/tidbcloud/AgentTeam/internal/api/daemon"
	"github.com/tidbcloud/AgentTeam/internal/api/im"
	"github.com/tidbcloud/AgentTeam/internal/api/im/fanout"
	"github.com/tidbcloud/AgentTeam/internal/api/llm"
	"github.com/tidbcloud/AgentTeam/internal/api/media"
	"github.com/tidbcloud/AgentTeam/internal/api/org"
	"github.com/tidbcloud/AgentTeam/internal/api/provider"
	"github.com/tidbcloud/AgentTeam/internal/api/public"
	"github.com/tidbcloud/AgentTeam/internal/api/schedule"
	"github.com/tidbcloud/AgentTeam/internal/api/serviceaccount"
	"github.com/tidbcloud/AgentTeam/internal/api/task"
	"github.com/tidbcloud/AgentTeam/internal/api/team"
	"github.com/tidbcloud/AgentTeam/internal/api/teamworkflow"
	apitestkit "github.com/tidbcloud/AgentTeam/internal/api/testkit"
	apitrigger "github.com/tidbcloud/AgentTeam/internal/api/trigger"
	"github.com/tidbcloud/AgentTeam/internal/api/webagent"
	"github.com/tidbcloud/AgentTeam/internal/api/workflow/binding"
	"github.com/tidbcloud/AgentTeam/internal/api/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/api/workflow/plan"
	"github.com/tidbcloud/AgentTeam/internal/api/workflow/run"
	"github.com/tidbcloud/AgentTeam/internal/api/workspace"
	aprruntime "github.com/tidbcloud/AgentTeam/internal/apr/runtime"
	"github.com/tidbcloud/AgentTeam/internal/archive"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/coredb/testkit"
	"github.com/tidbcloud/AgentTeam/internal/mailer"
	"github.com/tidbcloud/AgentTeam/internal/router"
	"github.com/tidbcloud/AgentTeam/internal/runtimecontrol"
	"github.com/tidbcloud/AgentTeam/internal/schedule/manage"
	"github.com/tidbcloud/AgentTeam/internal/service/agentruntime"
	"github.com/tidbcloud/AgentTeam/internal/service/appruntime"
	"github.com/tidbcloud/AgentTeam/internal/service/authflow"
	messageservice "github.com/tidbcloud/AgentTeam/internal/service/message"
	serviceaccountservice "github.com/tidbcloud/AgentTeam/internal/service/serviceaccount"
	serviceteamworkflow "github.com/tidbcloud/AgentTeam/internal/service/teamworkflow"
	workspaceservice "github.com/tidbcloud/AgentTeam/internal/service/workspace"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
	"github.com/tidbcloud/AgentTeam/internal/store/organization"
	"github.com/tidbcloud/AgentTeam/internal/store/taskboard"
	bindingstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/binding"
	calendarstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/calendar"
	invocationstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/invocation"
	planstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/plan"
	runstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	storeworkspace "github.com/tidbcloud/AgentTeam/internal/store/workspace"
	"github.com/tidbcloud/AgentTeam/internal/trigger"
	"github.com/tidbcloud/AgentTeam/internal/versioning"
	"github.com/tidbcloud/AgentTeam/internal/workflow/bindingmanage"
	"github.com/tidbcloud/AgentTeam/internal/workflow/calendar"
	"github.com/tidbcloud/AgentTeam/internal/workflow/invocationmanage"
	"github.com/tidbcloud/AgentTeam/internal/workflow/orchestrator"
	"github.com/tidbcloud/AgentTeam/internal/workflow/planmanage"
	"github.com/tidbcloud/AgentTeam/internal/workflow/runmanage"
)

func TestWorkflowCLIAgentRoundTrip(t *testing.T) {
	for _, command := range []string{"curl", "git", "go", "node"} {
		requireCommand(t, command)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
	defer cancel()

	tidbRuntime := startTiDBRuntime(t)
	testServer := startTestServer(t, ctx, tidbRuntime.databaseDSN)
	client := newAPIClient(t, testServer.baseURL)

	registered := mustDoJSON[registerResponse](ctx, client, http.MethodPost, "/api/v1/auth/register", http.StatusOK, map[string]string{
		"email":    "workflow-cli-e2e@example.com",
		"password": "super-secret-password",
		"name":     "workflow cli e2e",
	})
	client.accessToken = registered.AccessToken
	workspaces := mustDoJSON[[]apiWorkspace](ctx, client, http.MethodGet, "/api/v1/workspaces", http.StatusOK, nil)
	require.NotEmpty(t, workspaces)
	workspaceID := workspaces[0].ID
	client.workspaceID = workspaceID

	createdMachine := mustDoJSON[createMachineResponse](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/machines", workspaceID), http.StatusOK, map[string]string{
		"name": "workflow-cli-e2e-machine",
	})
	_ = startAgentTeamDaemon(t, ctx, testServer.baseURL, createdMachine.APIKey)
	waitForMachineConnected(t, ctx, client, workspaceID, createdMachine.Machine.ID)

	createdAgent := mustDoJSON[apiAgent](ctx, client, http.MethodPost, workspaceAgentsPath(workspaceID), http.StatusOK, map[string]any{
		"name":        "workflow-cli-app",
		"description": "Workflow CLI end-to-end app",
		"runtime":     "codex",
		"machineId":   createdMachine.Machine.ID,
	})
	channel := mustDoJSON[apiChannel](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels", workspaceID), http.StatusOK, map[string]string{
		"name":        "workflow-cli",
		"description": "Workflow CLI end-to-end verification",
	})
	mustDoJSON[map[string]any](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels/%s/members", workspaceID, channel.ID), http.StatusOK, map[string]string{
		"agentId":    createdAgent.ID,
		"visibility": "public",
	})
	mustDoJSON[apiMessage](ctx, client, http.MethodPost, scopeMessagesPath(workspaceID, channel.ID), http.StatusOK, map[string]string{
		"scopeId": channel.ID,
		"content": fmt.Sprintf("@%s Run the complete Workflow CLI verification. [daemon-e2e-workflow-cli]", createdAgent.Name),
	})

	require.EventuallyWithT(t, func(collect *assert.CollectT) {
		var planArchivedAt *time.Time
		var planRevision, draftRevision, versionCount int64
		err := testServer.pool.QueryRow(ctx, `
SELECT p.archived_at, p.revision, d.revision,
       (SELECT COUNT(*) FROM workflow_plan_versions v WHERE v.workspace_id=p.workspace_id AND v.plan_id=p.id)
FROM workflow_plans p
JOIN workflow_plan_drafts d ON d.workspace_id=p.workspace_id AND d.plan_id=p.id
WHERE p.workspace_id=? AND p.created_by_actor_type='agent' AND p.created_by_actor_id=?
`, workspaceID, createdAgent.ID).Scan(&planArchivedAt, &planRevision, &draftRevision, &versionCount)
		require.NoError(collect, err)
		require.NotNil(collect, planArchivedAt)
		require.GreaterOrEqual(collect, planRevision, int64(3))
		require.Equal(collect, int64(2), draftRevision)
		require.Equal(collect, int64(1), versionCount)

		var bindingState, sourceStatus string
		var bindingRevision, latestRevisionNumber int64
		err = testServer.pool.QueryRow(ctx, `
SELECT b.state, b.revision, r.revision_number, s.status
FROM workflow_bindings b
JOIN workflow_binding_revisions r ON r.id=b.latest_revision_id AND r.workspace_id=b.workspace_id
JOIN workflow_binding_sources s ON s.binding_id=b.id AND s.workspace_id=b.workspace_id
WHERE b.workspace_id=? AND b.created_by_actor_type='agent' AND b.created_by_actor_id=?
`, workspaceID, createdAgent.ID).Scan(&bindingState, &bindingRevision, &latestRevisionNumber, &sourceStatus)
		require.NoError(collect, err)
		require.Equal(collect, "archived", bindingState)
		require.GreaterOrEqual(collect, bindingRevision, int64(5))
		require.Equal(collect, int64(2), latestRevisionNumber)
		require.Equal(collect, "disabled", sourceStatus)

		var invocations, runs int64
		err = testServer.pool.QueryRow(ctx, `
SELECT COUNT(*), COUNT(r.id)
FROM workflow_invocations i
LEFT JOIN workflow_run_instances r ON r.workspace_id=i.workspace_id AND r.invocation_id=i.id
WHERE i.workspace_id=? AND i.invoked_by_actor_type='agent' AND i.invoked_by_actor_id=?
`, workspaceID, createdAgent.ID).Scan(&invocations, &runs)
		require.NoError(collect, err)
		require.Equal(collect, int64(1), invocations)
		require.Equal(collect, int64(1), runs)
	}, 90*time.Second, 500*time.Millisecond)
}

func TestDaemonProtocolRoundTrip(t *testing.T) {
	for _, command := range []string{"curl", "git", "go", "node"} {
		requireCommand(t, command)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 8*time.Minute)
	defer cancel()

	tidbRuntime := startTiDBRuntime(t)
	testServer := startTestServer(t, ctx, tidbRuntime.databaseDSN)
	client := newAPIClient(t, testServer.baseURL)

	registered := mustDoJSON[registerResponse](ctx, client, http.MethodPost, "/api/v1/auth/register", http.StatusOK, map[string]string{
		"email":    "daemon-e2e@example.com",
		"password": "super-secret-password",
		"name":     "daemon e2e",
	})
	require.NotEmpty(t, registered.AccessToken)
	require.NotEmpty(t, registered.RefreshToken)
	client.accessToken = registered.AccessToken

	workspaces := mustDoJSON[[]apiWorkspace](ctx, client, http.MethodGet, "/api/v1/workspaces", http.StatusOK, nil)
	require.NotEmpty(t, workspaces)

	workspaceID := workspaces[0].ID
	require.NotEmpty(t, workspaceID)
	client.workspaceID = workspaceID

	channels := mustDoJSON[[]apiChannel](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/channels", workspaceID), http.StatusOK, nil)
	allChannelID := channelIDByName(t, channels, "all")
	require.NotEmpty(t, allChannelID)

	opsChannel := mustDoJSON[apiChannel](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels", workspaceID), http.StatusOK, map[string]string{
		"name":        "ops",
		"description": "Operations channel for daemon e2e",
	})

	createdMachine := mustDoJSON[createMachineResponse](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/machines", workspaceID), http.StatusOK, map[string]string{
		"name": "daemon-e2e-machine",
	})
	require.NotEmpty(t, createdMachine.Machine.ID)
	require.NotEmpty(t, createdMachine.APIKey)

	daemon := startAgentTeamDaemon(t, ctx, testServer.baseURL, createdMachine.APIKey)
	waitForMachineConnected(t, ctx, client, workspaceID, createdMachine.Machine.ID)

	createdAgent := mustDoJSON[apiAgent](ctx, client, http.MethodPost, workspaceAgentsPath(workspaceID), http.StatusOK, map[string]any{
		"name":        "ops-app",
		"description": "daemon e2e app",
		"runtime":     "codex",
		"machineId":   createdMachine.Machine.ID,
	})
	require.NotEmpty(t, createdAgent.ID)
	makeAgentPublicChannelParticipantForDaemonE2ETest(t, ctx, testServer.pool, workspaceID, allChannelID, createdAgent.ID)

	usage := mustDoJSON[storeworkspace.WorkspaceUsage](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/usage", workspaceID), http.StatusOK, nil)
	require.EqualValues(t, 1, usage.MachineCount)
	require.EqualValues(t, 1, usage.AgentCount)
	require.EqualValues(t, 1, usage.ChannelCount)

	batchChannel := mustDoJSON[apiChannel](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels", workspaceID), http.StatusOK, map[string]string{
		"name":        "batch-claim",
		"description": "Task batch-claim channel for daemon e2e",
	})
	mustDoJSON[map[string]any](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels/%s/members", workspaceID, batchChannel.ID), http.StatusOK, map[string]string{
		"agentId":    createdAgent.ID,
		"visibility": "public",
	})
	createdBatchTasks := mustDoJSON[apiTaskPage](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels/%s/tasks", workspaceID, batchChannel.ID), http.StatusOK, map[string]any{
		"tasks": []map[string]string{
			{"title": "Batch-claim task one"},
			{"title": "Batch-claim task two"},
		},
	})
	require.Len(t, createdBatchTasks.Tasks, 2)
	require.EqualValues(t, 1, createdBatchTasks.Tasks[0].TaskNumber)
	require.EqualValues(t, 2, createdBatchTasks.Tasks[1].TaskNumber)

	mustDoJSON[apiMessage](ctx, client, http.MethodPost, scopeMessagesPath(workspaceID, batchChannel.ID), http.StatusOK, map[string]string{
		"scopeId": batchChannel.ID,
		"content": fmt.Sprintf("@%s Claim the two existing tasks. [daemon-e2e-batch-claim]", createdAgent.Name),
	})
	batchReply := waitForChannelReplyContaining(t, ctx, client, batchChannel.ID, "#t1")
	require.Contains(t, strings.ToLower(batchReply.Content), "claimed")
	require.Contains(t, batchReply.Content, "#t2")
	waitForClaimedTasks(t, ctx, client, batchChannel.ID, []int64{1, 2}, createdAgent.ID)
	waitForWorkspaceFile(t, ctx, client, createdAgent.ID, "run-summary.json")

	mustDoJSON[map[string]any](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels/%s/members", workspaceID, opsChannel.ID), http.StatusOK, map[string]string{
		"agentId":    createdAgent.ID,
		"visibility": "public",
	})

	threadPrompt := mustDoJSON[apiMessage](ctx, client, http.MethodPost, scopeMessagesPath(workspaceID, opsChannel.ID), http.StatusOK, map[string]string{
		"scopeId": opsChannel.ID,
		"content": fmt.Sprintf("@%s Reply in a thread with an attachment. [daemon-e2e-thread-attachment]", createdAgent.Name),
	})
	require.NotEmpty(t, threadPrompt.ID)
	threadSummary := waitForReplyThreadSummary(t, ctx, client, opsChannel.ID, threadPrompt.ID)
	require.NotEmpty(t, threadSummary.ThreadChannelID)
	require.EqualValues(t, 1, threadSummary.ReplyCount)
	require.Contains(t, threadSummary.ParticipantIDs, createdAgent.ID)

	threadReply := waitForAttachmentOnlyReply(t, ctx, client, threadSummary.ThreadChannelID, createdAgent.ID)
	require.Empty(t, threadReply.Content)
	require.Len(t, threadReply.Attachments, 1)
	require.Equal(t, "thread-reply.png", threadReply.Attachments[0].Filename)

	mustDoJSON[apiMessage](ctx, client, http.MethodPost, scopeMessagesPath(workspaceID, opsChannel.ID), http.StatusOK, map[string]string{
		"scopeId": opsChannel.ID,
		"content": fmt.Sprintf("@%s Please triage incident 42 and report back.", createdAgent.Name),
	})

	firstReply := waitForChannelReply(t, ctx, client, opsChannel.ID, createdAgent.ID)
	require.Contains(t, firstReply.Content, "processed #ops with #ops #t1")

	waitForCompletedTask(t, ctx, client, opsChannel.ID, 1, createdAgent.ID)

	mustDoJSON[apiMessage](ctx, client, http.MethodPost, scopeMessagesPath(workspaceID, opsChannel.ID), http.StatusOK, map[string]string{
		"scopeId": opsChannel.ID,
		"content": fmt.Sprintf("@%s Please follow up after sleeping and report back.", createdAgent.Name),
	})
	var followUpReply apiMessage
	require.Eventually(t, func() bool {
		page := mustDoJSON[apiMessagePage](ctx, client, http.MethodGet, scopeMessagesPathWithLimit(workspaceID, opsChannel.ID, 20), http.StatusOK, nil)
		for _, message := range page.Messages {
			if message.AuthorProfile.ActorType == "agent" && message.AuthorProfile.PrincipalID == createdAgent.ID && strings.Contains(message.Content, "processed #ops with #ops #t2") {
				followUpReply = message
				return true
			}
		}
		return false
	}, 90*time.Second, 500*time.Millisecond)
	require.Contains(t, followUpReply.Content, "processed #ops with #ops #t2")
	waitForCompletedTask(t, ctx, client, opsChannel.ID, 2, createdAgent.ID)

	rootFiles := waitForWorkspaceFile(t, ctx, client, createdAgent.ID, "run-summary.json")
	require.Contains(t, filenames(rootFiles), "run-summary.json")
	require.Contains(t, filenames(rootFiles), "notes")

	runSummaryBeforeReset := mustDoJSON[workspaceFileContent](ctx, client, http.MethodGet, workspaceAgentFileContentPath(workspaceID, createdAgent.ID, "run-summary.json"), http.StatusOK, nil)
	require.False(t, runSummaryBeforeReset.Binary)
	require.Contains(t, stringValue(runSummaryBeforeReset.Content), "#ops")
	runSummaryEntries := parseRunSummaryEntries(t, runSummaryBeforeReset)
	threadEntries := filterRunSummaryEntriesByReplyTarget(runSummaryEntries, threadSummaryKey(opsChannel.Name, threadPrompt.ID))
	require.Len(t, threadEntries, 1)
	require.Equal(t, "thread-reply.png", threadEntries[0].AttachmentFilename)
	require.Equal(t, "image/png", threadEntries[0].AttachmentDownloadContentType)
	require.Equal(t, 68, threadEntries[0].AttachmentDownloadSizeBytes)
	opsEntries := filterRunSummaryEntriesByReplyTarget(runSummaryEntries, "#ops")
	require.Len(t, opsEntries, 2)
	firstSessionID := strings.TrimSpace(opsEntries[0].SessionID)
	require.NotEmpty(t, firstSessionID)
	require.Equal(t, firstSessionID, strings.TrimSpace(opsEntries[1].SessionID))

	directoriesBeforeReset := mustDoJSON[apiMachineWorkspacePage](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/machines/%s/workspace-directories", workspaceID, createdMachine.Machine.ID), http.StatusOK, nil)
	require.Contains(t, directoryNames(directoriesBeforeReset.Workspaces), createdAgent.ID)

	mustDoJSON[map[string]any](ctx, client, http.MethodPost, workspaceAgentPath(workspaceID, createdAgent.ID, "/reset"), http.StatusOK, map[string]any{
		"clearWorkspace": true,
	})
	waitForWorkspaceEmpty(t, ctx, client, createdAgent.ID)
	waitForWorkspaceDirectoryMissing(t, ctx, client, workspaceID, createdMachine.Machine.ID, createdAgent.ID)

	dmChannel := mustDoJSON[dmChannel](ctx, client, http.MethodPost, fmt.Sprintf("/api/v1/workspaces/%s/channels/dm", workspaceID), http.StatusOK, map[string]string{
		"agentId": createdAgent.ID,
	})
	mustDoJSON[apiMessage](ctx, client, http.MethodPost, scopeMessagesPath(workspaceID, dmChannel.ID), http.StatusOK, map[string]string{
		"scopeId": dmChannel.ID,
		"content": "Ping after workspace reset.",
	})
	mustStartAgentOrAlreadyRunning(ctx, client, createdAgent.ID)

	secondReply := waitForChannelReply(t, ctx, client, dmChannel.ID, createdAgent.ID)
	require.Contains(t, secondReply.Content, "processed dm:@")
	require.Contains(t, secondReply.Content, "#all #t1")

	runSummaryAfterReset := mustDoJSON[workspaceFileContent](ctx, client, http.MethodGet, workspaceAgentFileContentPath(workspaceID, createdAgent.ID, "run-summary.json"), http.StatusOK, nil)
	require.False(t, runSummaryAfterReset.Binary)
	require.Contains(t, stringValue(runSummaryAfterReset.Content), "#all")
	require.NotContains(t, stringValue(runSummaryAfterReset.Content), "#ops #t1")
	runSummaryEntriesAfterReset := parseRunSummaryEntries(t, runSummaryAfterReset)
	require.Len(t, runSummaryEntriesAfterReset, 1)
	secondSessionID := strings.TrimSpace(runSummaryEntriesAfterReset[0].SessionID)
	require.NotEmpty(t, secondSessionID)
	require.NotEqual(t, firstSessionID, secondSessionID)

	directoriesAfterSecondRun := mustDoJSON[apiMachineWorkspacePage](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/machines/%s/workspace-directories", workspaceID, createdMachine.Machine.ID), http.StatusOK, nil)
	require.Contains(t, directoryNames(directoriesAfterSecondRun.Workspaces), createdAgent.ID)

	mustDoJSON[map[string]any](ctx, client, http.MethodDelete, fmt.Sprintf("/api/v1/workspaces/%s/machines/%s/workspace-directories/%s", workspaceID, createdMachine.Machine.ID, createdAgent.ID), http.StatusOK, nil)
	waitForWorkspaceDirectoryMissing(t, ctx, client, workspaceID, createdMachine.Machine.ID, createdAgent.ID)

	t.Logf("agentteam daemon logs:\n%s", daemon.logs())
}

type apiClient struct {
	t           *testing.T
	baseURL     string
	httpClient  *http.Client
	accessToken string
	workspaceID string
}

func newAPIClient(t *testing.T, baseURL string) *apiClient {
	t.Helper()
	return &apiClient{
		t:          t,
		baseURL:    baseURL,
		httpClient: &http.Client{Timeout: 15 * time.Second},
	}
}

func mustDoJSON[T any](ctx context.Context, client *apiClient, method string, path string, expectedStatus int, body any) T {
	client.t.Helper()

	var requestBody io.Reader
	if body != nil {
		payload, err := json.Marshal(body)
		require.NoError(client.t, err)
		requestBody = bytes.NewReader(payload)
	}

	request, err := http.NewRequestWithContext(ctx, method, client.baseURL+path, requestBody)
	require.NoError(client.t, err)
	if body != nil {
		request.Header.Set("Content-Type", "application/json")
	}
	if client.accessToken != "" {
		request.Header.Set("Authorization", "Bearer "+client.accessToken)
	}
	if client.workspaceID != "" {
		request.Header.Set("X-Workspace-Id", client.workspaceID)
	}

	response, err := client.httpClient.Do(request)
	require.NoError(client.t, err)
	defer response.Body.Close()

	payload, err := io.ReadAll(response.Body)
	require.NoError(client.t, err)
	require.Equalf(client.t, expectedStatus, response.StatusCode, "%s %s: %s", method, path, string(payload))

	var decoded T
	if len(bytes.TrimSpace(payload)) == 0 {
		return decoded
	}
	require.NoError(client.t, json.Unmarshal(payload, &decoded), "%s %s: %s", method, path, string(payload))
	return decoded
}

func workspaceAgentsPath(workspaceID string) string {
	return fmt.Sprintf("/api/v1/workspaces/%s/agents", url.PathEscape(workspaceID))
}

func workspaceAgentPath(workspaceID string, agentID string, suffix string) string {
	return fmt.Sprintf("%s/%s%s", workspaceAgentsPath(workspaceID), url.PathEscape(agentID), suffix)
}

func workspaceAgentFilesPath(workspaceID string, agentID string) string {
	return workspaceAgentPath(workspaceID, agentID, "/files")
}

func workspaceAgentFileContentPath(workspaceID string, agentID string, path string) string {
	return fmt.Sprintf("%s/content?path=%s", workspaceAgentFilesPath(workspaceID, agentID), url.QueryEscape(path))
}

func mustStartAgentOrAlreadyRunning(ctx context.Context, client *apiClient, agentID string) {
	client.t.Helper()

	path := workspaceAgentPath(client.workspaceID, agentID, "/start")
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, client.baseURL+path, nil)
	require.NoError(client.t, err)
	if client.accessToken != "" {
		request.Header.Set("Authorization", "Bearer "+client.accessToken)
	}
	if client.workspaceID != "" {
		request.Header.Set("X-Workspace-Id", client.workspaceID)
	}

	response, err := client.httpClient.Do(request)
	require.NoError(client.t, err)
	defer response.Body.Close()

	payload, err := io.ReadAll(response.Body)
	require.NoError(client.t, err)
	if response.StatusCode == http.StatusOK {
		return
	}
	if response.StatusCode == http.StatusConflict && strings.Contains(string(payload), "agent is still running") {
		return
	}
	require.Equalf(client.t, http.StatusOK, response.StatusCode, "%s %s: %s", http.MethodPost, path, string(payload))
}

func scopeMessagesPath(workspaceID string, scopeID string) string {
	return fmt.Sprintf(
		"/api/v1/workspaces/%s/scopes/%s/messages",
		url.PathEscape(workspaceID),
		url.PathEscape(scopeID),
	)
}

func scopeMessagesPathWithLimit(workspaceID string, scopeID string, limit int) string {
	return fmt.Sprintf("%s?limit=%d", scopeMessagesPath(workspaceID, scopeID), limit)
}

type registerResponse struct {
	AccessToken  string  `json:"accessToken"`
	RefreshToken string  `json:"refreshToken"`
	User         apiUser `json:"user"`
}

type createMachineResponse struct {
	Machine apiMachine `json:"machine"`
	APIKey  string     `json:"apiKey"`
}

type apiUser struct {
	ID string `json:"id"`
}

type apiWorkspace struct {
	ID   string `json:"id"`
	Name string `json:"name"`
}

type apiChannel struct {
	ID          string `json:"id"`
	Name        string `json:"name"`
	Description string `json:"description,omitempty"`
	Joined      bool   `json:"joined,omitempty"`
}

type apiMachine struct {
	ID       string   `json:"id"`
	Status   string   `json:"status"`
	Runtimes []string `json:"runtimes"`
}

type apiMachinesList struct {
	Machines            []apiMachine `json:"machines"`
	LatestDaemonVersion string       `json:"latestDaemonVersion"`
}

func (list *apiMachinesList) UnmarshalJSON(data []byte) error {
	trimmed := bytes.TrimSpace(data)
	if len(trimmed) == 0 {
		return nil
	}

	switch trimmed[0] {
	case '{':
		var payload struct {
			Machines            []apiMachine `json:"machines"`
			LatestDaemonVersion string       `json:"latestDaemonVersion"`
		}
		if err := json.Unmarshal(trimmed, &payload); err != nil {
			return err
		}
		list.Machines = payload.Machines
		list.LatestDaemonVersion = payload.LatestDaemonVersion
		return nil
	case '[':
		var machines []apiMachine
		if err := json.Unmarshal(trimmed, &machines); err != nil {
			return err
		}
		list.Machines = machines
		list.LatestDaemonVersion = ""
		return nil
	default:
		return fmt.Errorf("unexpected machines response: %s", string(trimmed))
	}
}

type apiAgent struct {
	ID     string `json:"id"`
	Name   string `json:"name"`
	Status string `json:"status"`
}

type apiAttachment struct {
	ID       string `json:"id"`
	Filename string `json:"filename"`
}

type apiAuthorProfile struct {
	ActorType   string `json:"actorType"`
	PrincipalID string `json:"principalId,omitempty"`
	MemberID    string `json:"memberId,omitempty"`
	TargetID    string `json:"targetId,omitempty"`
	DisplayName string `json:"displayName"`
}

type apiMessage struct {
	ID            string           `json:"id"`
	Seq           int64            `json:"seq"`
	AuthorProfile apiAuthorProfile `json:"authorProfile"`
	Content       string           `json:"content"`
	Attachments   []apiAttachment  `json:"attachments,omitempty"`
}

type apiMessagePage struct {
	Messages []apiMessage `json:"messages"`
}

type fakeCodexRunSummaryEntry struct {
	SessionID                     string `json:"sessionId"`
	ReplyTarget                   string `json:"replyTarget"`
	AttachmentFilename            string `json:"attachmentFilename,omitempty"`
	AttachmentDownloadContentType string `json:"attachmentDownloadContentType,omitempty"`
	AttachmentDownloadSizeBytes   int    `json:"attachmentDownloadSizeBytes,omitempty"`
}

type apiTask struct {
	TaskNumber      int64             `json:"taskNumber"`
	Status          string            `json:"status"`
	AssigneeProfile *apiAuthorProfile `json:"assigneeProfile,omitempty"`
}

type apiTaskPage struct {
	Tasks []apiTask `json:"tasks"`
	Rows  []apiTask `json:"rows"`
}

type apiMachineWorkspace struct {
	DirectoryName string `json:"directoryName"`
}

type apiMachineWorkspacePage struct {
	Workspaces []apiMachineWorkspace `json:"workspaces"`
}

type dmChannel struct {
	ID string `json:"id"`
}

type workspaceTreeResponse struct {
	Files []workspaceFile `json:"files"`
}

type workspaceFile struct {
	Name string `json:"name"`
}

type workspaceFileContent struct {
	Content *string `json:"content"`
	Binary  bool    `json:"binary"`
}

type tidbRuntime struct {
	databaseDSN string
}

func startTiDBRuntime(t *testing.T) tidbRuntime {
	t.Helper()

	return tidbRuntime{
		databaseDSN: startE2EInitializedTiDB(t),
	}
}

type testServerRuntime struct {
	baseURL string
	server  *httptest.Server
	pool    coredb.DB
}

func startTestServer(t *testing.T, ctx context.Context, databaseDSN string) testServerRuntime {
	t.Helper()

	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
	sqlDB := testkit.OpenTiDB(t, databaseDSN)
	corePool := coredb.WrapSQLDB(sqlDB)

	pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	require.NoError(t, sqlDB.PingContext(pingCtx))

	archiveStorage, err := archive.NewFilesystem(t.TempDir())
	require.NoError(t, err)

	dataStore := apitestkit.NewWithOptions(corePool, archiveStorage, 30, apitestkit.StoreOptions{
		AppCredentialStorageKey: []byte("0123456789abcdef0123456789abcdef"),
	})
	appStore := dataStore.App()
	appRunnerService := appruntime.NewRunnerService(appruntime.RunnerServiceOptions{Store: dataStore.Graph.ServiceActor, Authority: appStore})
	workspaceStore := dataStore.Workspace()
	channelStore := dataStore.Channel()
	privateAgentStore := dataStore.PrivateAgents()
	identityStore := dataStore.Identity()
	taskBoard := taskboard.New(corePool)
	server := httptest.NewUnstartedServer(http.NotFoundHandler())
	baseURL := "http://" + server.Listener.Addr().String()
	noOpMailer := mailer.NewNoop(logger, baseURL)
	agentRuntime := agentruntime.New(agentruntime.Dependencies{
		Agents:           dataStore.Agent(),
		Messages:         dataStore.Message(),
		Recovery:         dataStore.Rewind(),
		WebAgents:        dataStore.App(),
		WebRecovery:      dataStore.Rewind(),
		IsRuntimeOffline: func(err error) bool { return errors.Is(err, runtimecontrol.ErrMachineOffline) },
	})
	replyRootPreloader := fanout.NewScopeReplyThreadRootPreloader(dataStore.Message())
	imStreamHub := im.NewStreamEventHub()
	imStreamWriter := im.NewPublishingStreamEventWriter(dataStore.IM(), imStreamHub)
	liveEmitter := im.NewLiveEventEmitter(imStreamHub, appStore, logger)
	daemonHub := daemon.New(logger, daemon.NewStore(dataStore.Agent(), dataStore.Rewind(), agentRuntime.PrepareAgentStart), baseURL, liveEmitter, dataStore.AgentOps())
	agentRuntime.SetRuntimeControl(daemonHub)
	deliveryFanout := fanout.NewDeliveryFanout(daemonHub, imStreamWriter, logger).
		WithReplyThreadRootPreloader(replyRootPreloader).
		WithScopeFanoutMarker(appStore).
		WithTaskScopeLifecycleGuard(dataStore.Message())
	agentAPI := agent.New(logger, dataStore.AgentStorePortsDeps(), dataStore.Attachment(), taskBoard, daemonHub, deliveryFanout)
	organizationService := organization.NewWithDB(corePool, apitestkit.NewOrganizationCoreBridge(corePool, dataStore))
	workspaceFlow := workspaceservice.New(workspaceservice.Dependencies{
		Workspaces:      workspaceStore,
		WorkspaceAdmin:  appStore,
		Machines:        appStore,
		MachineTokens:   dataStore.Agent(),
		MiniappPairings: identityStore,
	})
	triggerService := trigger.New(logger, corePool, appStore, trigger.ServiceOptions{})
	agentAPI.TriggerService = triggerService
	scheduleService := manage.New(logger, corePool, appStore, manage.ServiceOptions{})
	agentAPI.ScheduleService = scheduleService
	workflowPlanService := planmanage.New(planstore.New(corePool))
	workflowBindingService := bindingmanage.New(
		bindingstore.New(corePool),
		workflowPlanService,
		bindingmanage.WithAgentOwnerVerifier(appStore),
	)
	workflowInvocationService := invocationmanage.New(invocationstore.New(corePool))
	workflowRunStore := runstore.New(corePool)
	require.NoError(t, workflowRunStore.Validate())
	workflowRunService := runmanage.New(workflowRunStore)
	workflowCalendarService := calendar.New(calendarstore.New(corePool))
	workflowRunCallbacks, err := orchestrator.NewCallbackService(workflowRunStore, orchestrator.CallbackServiceOptions{
		ScopeAuthorizer: appStore,
	})
	require.NoError(t, err)
	agentAPI.WorkflowPlans = workflowPlanService
	agentAPI.WorkflowBindings = workflowBindingService
	agentAPI.WorkflowBindingOwners = appStore
	agentAPI.WorkflowBindingWorkspace = workspaceStore
	agentAPI.WorkflowInvocations = workflowInvocationService
	agentAPI.WorkflowInvocationOwners = appStore
	agentAPI.WorkflowInvocationWorkspace = workspaceStore
	agentAPI.WorkflowRuns = workflowRunService
	agentAPI.WorkflowRecovery = workflowRunStore
	agentAPI.WorkflowRunCallbacks = workflowRunCallbacks
	teamWorkflows := serviceteamworkflow.New(corePool)
	agentAPI.TeamWorkflowPlans = teamWorkflows
	teamWorkflowBrowserService := teamworkflow.NewStoreService(corePool, teamWorkflows)
	serviceAccountFlow := serviceaccountservice.New(corePool)
	messagingService := messageservice.New(messageservice.Dependencies{
		Sessions:   identityStore,
		Workspaces: workspaceFlow,
		Creation:   appStore,
		Queries:    appStore,
		Search:     dataStore.Message(),
		Logger:     logger,
	})
	coreDBAdmin := coredb.NewAdmin()
	adminDB, err := coreDBAdmin.Open(ctx, databaseDSN, 5*time.Second)
	require.NoError(t, err)
	t.Cleanup(adminDB.Close)
	aprDispatcher := aprruntime.NewDispatcher()
	t.Cleanup(aprDispatcher.Stop)
	routerDependencies := router.Dependencies{
		Logger: logger,
		Health: router.HealthDependencies{
			CoreDB:      adminDB,
			CoreDBAdmin: coreDBAdmin,
		},
		Agent: agentAPI,
		Daemon: router.DaemonDependencies{
			Hub:           daemonHub,
			Auth:          dataStore.Agent(),
			SessionIssuer: identityStore,
		},
		API: router.APIDependencies{
			Auth: auth.Dependencies{
				Store: identityStore,
				AuthFlow: authflow.New(authflow.Dependencies{
					RegistrationAllowlist: identityStore,
					WorkspaceInvites:      appStore,
					Organizations:         organizationService,
				}),
				Organizations:        organizationService,
				Mailer:               noOpMailer,
				RegistrationPolicy:   auth.NewRegistrationPolicy(auth.RegistrationModeOpen, nil),
				VerificationTokenTTL: 24 * time.Hour,
				SessionTTL:           7 * 24 * time.Hour,
			},
			Admin: admin.Dependencies{
				AuthStore:             identityStore,
				AdminStore:            dataStore.Admin(),
				RegistrationAllowlist: identityStore,
			},
			Workspace: workspace.Dependencies{
				AuthStore:           identityStore,
				WorkspaceService:    workspaceFlow,
				PlanStore:           dataStore.Plan(),
				Organizations:       organizationService,
				Mailer:              noOpMailer,
				MachineRuntime:      daemonHub,
				LatestDaemonVersion: versioning.NewStaticProvider(versioning.ServerVersion()),
			},
			Org: org.Dependencies{
				Store:         dataStore,
				Organizations: organizationService,
				PlanStore:     dataStore.Plan(),
				Mailer:        noOpMailer,
			},
			Public: public.Dependencies{
				AuthStore:                  identityStore,
				MessageSearchStore:         dataStore.Message(),
				ContentReports:             dataStore.Reporting(),
				OpenRegistration:           true,
				RegistrationMode:           auth.RegistrationModeOpen,
				DaemonConnectBaseURL:       baseURL,
				MeshXKnowledgeBasesEnabled: true,
			},
			Media: media.Dependencies{
				AuthStore:            dataStore,
				WorkspaceStore:       dataStore,
				AgentTokenStore:      dataStore.Agent(),
				AgentStore:           appStore,
				AttachmentStore:      dataStore.Attachment(),
				PersonalStickerStore: dataStore.Attachment(),
			},
			IM: im.Dependencies{
				SessionAuth:              identityStore,
				WorkspaceStore:           workspaceFlow,
				Logger:                   logger,
				Messages:                 messagingService,
				MessageQueries:           messagingService,
				Channels:                 channelStore,
				ChannelAccess:            channelStore,
				ChannelMembers:           appStore,
				MessageSearch:            dataStore.Message(),
				HumanMessages:            messagingService,
				ScopeSnapshotTaskBoard:   im.NewScopeSnapshotTaskBoard(taskBoard),
				AgentDeliveries:          appStore,
				AgentNotifier:            daemonHub,
				ReplyThreadRootPreloader: replyRootPreloader,
				ScopeFanoutMarker:        appStore,
				TaskScopeLifecycle:       dataStore.Message(),
				RewindStore:              dataStore.Rewind(),
				RewindHub:                daemonHub,
				DMs:                      appStore,
				OwnerContexts:            appStore,
				ChannelNotifications:     privateAgentStore,
				ThreadNotifications:      privateAgentStore,
				Conversations:            dataStore.Message(),
				Reactions:                dataStore.Reaction(),
				Pins:                     dataStore.Message(),
				StreamEvents:             dataStore.IM(),
				StreamWriter:             imStreamWriter,
				StreamSubscriptions:      imStreamHub,
				StreamCursors:            dataStore.IM(),
				ConversationIndex:        dataStore.IM(),
				VisibleScopes:            message.NewSQLVisibleScopeAuthorizer(corePool),
			},
			Channel: channel.Dependencies{
				AuthStore:         dataStore,
				WorkspaceStore:    dataStore,
				Channels:          channelStore,
				AppStore:          appStore,
				ThreadStore:       dataStore.Message(),
				ThreadReviewStore: dataStore.Review(),
			},
			Task: task.Dependencies{
				AuthStore:                   dataStore,
				WorkspaceStore:              dataStore,
				TaskBoard:                   taskBoard,
				ReviewStore:                 dataStore.Review(),
				Store:                       task.NewTaskParityStoreAdapter(appStore, privateAgentStore),
				MessageFanout:               deliveryFanout,
				ConversationReadProjector:   im.NewTaskConversationProjector(dataStore.Message(), imStreamWriter, nil, liveEmitter),
				ConversationPolicyProjector: im.NewTaskConversationProjector(dataStore.Message(), imStreamWriter, nil, liveEmitter),
				Logger:                      logger,
			},
			Team: team.Dependencies{
				AuthStore:      identityStore,
				WorkspaceStore: workspaceFlow,
				Teams:          dataStore.Team(),
			},
			TeamWorkflow: teamworkflow.Dependencies{
				AuthStore:      identityStore,
				WorkspaceStore: workspaceStore,
				Service:        teamWorkflowBrowserService,
			},
			Schedule: schedule.Dependencies{
				AuthStore:       dataStore,
				WorkspaceStore:  dataStore,
				ScheduleService: scheduleService,
			},
			ServiceAccount: serviceaccount.Dependencies{
				SessionAuth:             identityStore,
				WorkspaceStore:          workspaceStore,
				Service:                 serviceAccountFlow,
				PendingRequestProjector: dataStore.ApprovalAppProjector(),
			},
			Trigger: apitrigger.Dependencies{
				AuthStore:      dataStore,
				WorkspaceStore: dataStore,
				TriggerService: triggerService,
			},
			WorkflowBinding: binding.Dependencies{
				AuthStore:      identityStore,
				WorkspaceStore: workspaceStore,
				Service:        workflowBindingService,
			},
			WorkflowInvocation: invocation.Dependencies{
				AuthStore:      identityStore,
				WorkspaceStore: workspaceStore,
				Service:        workflowInvocationService,
			},
			WorkflowPlan: plan.Dependencies{
				AuthStore:      identityStore,
				WorkspaceStore: workspaceStore,
				Service:        workflowPlanService,
			},
			WorkflowRun: run.Dependencies{
				AuthStore:      identityStore,
				WorkspaceStore: workspaceStore,
				Service:        workflowRunService,
				Calendar:       workflowCalendarService,
			},
			App: app.Dependencies{
				SessionAuth:                     identityStore,
				WorkspaceStore:                  workspaceFlow,
				AppStore:                        appStore,
				AppRuntimeStore:                 appStore,
				AppRunnerService:                appRunnerService,
				FollowupDeliverer:               deliveryFanout,
				ServiceEventSubscriptionBlocker: dataStore.App(),
				Logger:                          logger,
			},
			Provider: provider.Dependencies{
				SessionAuth:      dataStore,
				WorkspaceStore:   dataStore,
				Store:            appStore.Provider(),
				GitHubAdminStore: appStore,
				GitHubConfig: provider.GitHubConfig{
					FlowStateKey: []byte("daemon-e2e-github-flow-key"),
					BaseURL:      "https://github.com",
					APIBaseURL:   "https://api.github.com",
				},
				ServiceEventSubscriptionBlocker: dataStore.App(),
				HTTPClient:                      &http.Client{Timeout: 10 * time.Second},
				PublicBaseURL:                   baseURL,
				Logger:                          logger,
			},
			LLM: llm.Dependencies{
				APRDispatcher: aprDispatcher,
				Logger:        logger,
			},
			WebAgent: webagent.Dependencies{
				AgentStore:      appStore,
				ModelProfiles:   dataStore.Agent(),
				SessionAuth:     identityStore,
				WorkspaceStore:  workspaceStore,
				ReviewStore:     dataStore.Review(),
				RewindStore:     dataStore.Rewind(),
				RuntimeCommands: agentRuntime,
				Runtime:         daemonHub,
				Logger:          logger,
			},
		},
		LiveEmitter: liveEmitter,
	}
	require.NoError(t, routerDependencies.Validate())
	server.Config.Handler = router.NewRouter(routerDependencies)
	server.Start()
	t.Cleanup(server.Close)

	return testServerRuntime{
		baseURL: server.URL,
		server:  server,
		pool:    corePool,
	}
}

func makeAgentPublicChannelParticipantForDaemonE2ETest(t *testing.T, ctx context.Context, pool coredb.DB, workspaceID string, channelID string, agentID string) {
	t.Helper()

	apitestkit.MakeAgentPublicChannelParticipant(t, ctx, pool, workspaceID, channelID, agentID)
}

type daemonRuntime struct {
	command *exec.Cmd
	stdout  bytes.Buffer
	stderr  bytes.Buffer
}

func allocateAgentTeamDaemonPort(t *testing.T) int {
	t.Helper()

	listener, err := net.Listen("tcp", "127.0.0.1:0")
	require.NoError(t, err)
	defer listener.Close()

	address, ok := listener.Addr().(*net.TCPAddr)
	require.True(t, ok)
	return address.Port
}

func startAgentTeamDaemon(t *testing.T, ctx context.Context, serverURL string, machineAPIKey string) daemonRuntime {
	t.Helper()

	homeDir := t.TempDir()
	buildDir := filepath.Join(homeDir, "build")
	binDir := filepath.Join(homeDir, "bin")
	require.NoError(t, os.MkdirAll(buildDir, 0o755))
	require.NoError(t, os.MkdirAll(binDir, 0o755))
	port := allocateAgentTeamDaemonPort(t)

	daemonBinary := filepath.Join(buildDir, "agentteam")
	if runtime.GOOS == "windows" {
		daemonBinary += ".exe"
	}
	buildCmd := exec.CommandContext(ctx, "go", "build", "-o", daemonBinary, "./cmd/agentteam")
	buildCmd.Dir = filepath.Join("..", "desktop")
	buildOutput, err := buildCmd.CombinedOutput()
	require.NoErrorf(t, err, "build agentteam: %s", string(buildOutput))

	fakeCodexSource := filepath.Join("testdata", "fake-codex.mjs")
	fakeCodexScriptDestination := filepath.Join(binDir, "codex.mjs")
	fakeCodexDestination := filepath.Join(binDir, "codex")
	content, err := os.ReadFile(fakeCodexSource)
	require.NoError(t, err)
	require.NoError(t, os.WriteFile(fakeCodexScriptDestination, content, 0o755))
	require.NoError(t, os.WriteFile(fakeCodexDestination, []byte("#!/bin/sh\nexec node \""+fakeCodexScriptDestination+"\" \"$@\"\n"), 0o755))

	cmd := exec.CommandContext(
		ctx,
		daemonBinary,
		"daemon",
		"--connect",
		"agentteam",
		"--agentteam.server-url",
		serverURL,
		"--agentteam.api-key",
		machineAPIKey,
		"--port",
		fmt.Sprintf("%d", port),
	)
	cmd.Env = envWithOverrides(os.Environ(), map[string]string{
		"HOME":                  homeDir,
		"AGENTTEAM_CONFIG_DIR":       filepath.Join(homeDir, ".agentteam"),
		"AGENTTEAM_EXEC_BINARY_PATH": daemonBinary,
		"PATH":                  binDir + string(os.PathListSeparator) + os.Getenv("PATH"),
		"NO_COLOR":              "1",
		"FORCE_COLOR":           "0",
	})

	var runtime daemonRuntime
	runtime.command = cmd
	cmd.Stdout = &runtime.stdout
	cmd.Stderr = &runtime.stderr
	require.NoError(t, cmd.Start())
	t.Cleanup(func() {
		if cmd.ProcessState != nil && cmd.ProcessState.Exited() {
			return
		}
		_ = cmd.Process.Kill()
		_, _ = cmd.Process.Wait()
		if t.Failed() {
			t.Logf("agentteam daemon stdout:\n%s", runtime.stdout.String())
			t.Logf("agentteam daemon stderr:\n%s", runtime.stderr.String())
		}
	})

	return runtime
}

func (runtime daemonRuntime) logs() string {
	return strings.TrimSpace(runtime.stdout.String() + "\n" + runtime.stderr.String())
}

func envWithOverrides(base []string, overrides map[string]string) []string {
	if len(overrides) == 0 {
		return append([]string(nil), base...)
	}
	skip := make(map[string]struct{}, len(overrides))
	for key := range overrides {
		skip[canonicalEnvKey(key)] = struct{}{}
	}

	env := make([]string, 0, len(base)+len(overrides))
	for _, entry := range base {
		key, _, ok := strings.Cut(entry, "=")
		if !ok {
			continue
		}
		if _, exists := skip[canonicalEnvKey(key)]; exists {
			continue
		}
		env = append(env, entry)
	}
	for key, value := range overrides {
		env = append(env, key+"="+value)
	}
	return env
}

func canonicalEnvKey(key string) string {
	if runtime.GOOS == "windows" {
		return strings.ToUpper(key)
	}
	return key
}

func waitForMachineConnected(t *testing.T, ctx context.Context, client *apiClient, workspaceID string, machineID string) {
	t.Helper()

	require.Eventually(t, func() bool {
		response := mustDoJSON[apiMachinesList](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/machines", workspaceID), http.StatusOK, nil)
		for _, machine := range response.Machines {
			if machine.ID != machineID {
				continue
			}
			return machine.Status == "online" && contains(machine.Runtimes, "codex")
		}
		return false
	}, 90*time.Second, 500*time.Millisecond)
}

func waitForChannelReply(t *testing.T, ctx context.Context, client *apiClient, channelID string, agentID string) apiMessage {
	t.Helper()

	var reply apiMessage
	require.Eventually(t, func() bool {
		page := mustDoJSON[apiMessagePage](ctx, client, http.MethodGet, scopeMessagesPathWithLimit(client.workspaceID, channelID, 20), http.StatusOK, nil)
		for _, message := range page.Messages {
			if message.AuthorProfile.ActorType == "agent" && message.AuthorProfile.PrincipalID == agentID {
				reply = message
				return true
			}
		}
		return false
	}, 90*time.Second, 500*time.Millisecond)
	return reply
}

func waitForReplyThreadSummary(t *testing.T, ctx context.Context, client *apiClient, channelID string, parentMessageID string) message.ReplyThreadSummary {
	t.Helper()

	var summary message.ReplyThreadSummary
	require.Eventually(t, func() bool {
		response := mustDoJSON[map[string]message.ReplyThreadSummary](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/channels/%s/threads?limit=50", client.workspaceID, channelID), http.StatusOK, nil)
		candidate, ok := response[parentMessageID]
		if !ok {
			return false
		}
		if candidate.ThreadChannelID == "" || candidate.ReplyCount < 1 {
			return false
		}
		summary = candidate
		return true
	}, 90*time.Second, 500*time.Millisecond)
	return summary
}

func waitForAttachmentOnlyReply(t *testing.T, ctx context.Context, client *apiClient, channelID string, agentID string) apiMessage {
	t.Helper()

	var reply apiMessage
	var lastPage apiMessagePage
	require.Eventuallyf(t, func() bool {
		page := mustDoJSON[apiMessagePage](ctx, client, http.MethodGet, scopeMessagesPathWithLimit(client.workspaceID, channelID, 20), http.StatusOK, nil)
		lastPage = page
		for _, message := range page.Messages {
			if message.AuthorProfile.ActorType != "agent" || message.AuthorProfile.PrincipalID != agentID {
				continue
			}
			if message.Content != "" || len(message.Attachments) != 1 {
				continue
			}
			reply = message
			return true
		}
		return false
	}, 90*time.Second, 500*time.Millisecond, "last thread page: %+v", lastPage)
	return reply
}

func waitForChannelReplyContaining(t *testing.T, ctx context.Context, client *apiClient, channelID string, substring string) apiMessage {
	t.Helper()

	var reply apiMessage
	require.Eventually(t, func() bool {
		page := mustDoJSON[apiMessagePage](ctx, client, http.MethodGet, scopeMessagesPathWithLimit(client.workspaceID, channelID, 50), http.StatusOK, nil)
		for _, message := range page.Messages {
			if message.AuthorProfile.ActorType == "agent" && strings.Contains(message.Content, substring) {
				reply = message
				return true
			}
		}
		return false
	}, 90*time.Second, 500*time.Millisecond)
	return reply
}

func waitForCompletedTask(t *testing.T, ctx context.Context, client *apiClient, channelID string, taskNumber int64, agentID string) {
	t.Helper()

	require.Eventually(t, func() bool {
		response := mustDoJSON[apiTaskPage](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/channels/%s/tasks?limit=100", client.workspaceID, channelID), http.StatusOK, nil)
		for _, task := range response.Rows {
			if task.TaskNumber != taskNumber {
				continue
			}
			return task.Status == "done" && task.AssigneeProfile != nil &&
				task.AssigneeProfile.ActorType == "agent" && task.AssigneeProfile.PrincipalID == agentID
		}
		return false
	}, 90*time.Second, 500*time.Millisecond)
}

func waitForClaimedTasks(t *testing.T, ctx context.Context, client *apiClient, channelID string, taskNumbers []int64, agentID string) {
	t.Helper()

	expected := make(map[int64]struct{}, len(taskNumbers))
	for _, taskNumber := range taskNumbers {
		expected[taskNumber] = struct{}{}
	}

	require.Eventually(t, func() bool {
		response := mustDoJSON[apiTaskPage](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/channels/%s/tasks?limit=100", client.workspaceID, channelID), http.StatusOK, nil)
		matched := 0
		for _, task := range response.Rows {
			if _, ok := expected[task.TaskNumber]; !ok {
				continue
			}
			if task.Status != "todo" || task.AssigneeProfile == nil ||
				task.AssigneeProfile.ActorType != "agent" || task.AssigneeProfile.PrincipalID != agentID {
				return false
			}
			matched += 1
		}
		return matched == len(expected)
	}, 90*time.Second, 500*time.Millisecond)
}

func waitForWorkspaceFile(t *testing.T, ctx context.Context, client *apiClient, agentID string, fileName string) []workspaceFile {
	t.Helper()

	var files []workspaceFile
	require.Eventually(t, func() bool {
		response := mustDoJSON[workspaceTreeResponse](ctx, client, http.MethodGet, workspaceAgentFilesPath(client.workspaceID, agentID), http.StatusOK, nil)
		files = response.Files
		for _, file := range response.Files {
			if file.Name == fileName {
				return true
			}
		}
		return false
	}, 30*time.Second, 500*time.Millisecond)
	return files
}

func waitForWorkspaceEmpty(t *testing.T, ctx context.Context, client *apiClient, agentID string) {
	t.Helper()

	require.Eventually(t, func() bool {
		response := mustDoJSON[workspaceTreeResponse](ctx, client, http.MethodGet, workspaceAgentFilesPath(client.workspaceID, agentID), http.StatusOK, nil)
		return len(response.Files) == 0
	}, 30*time.Second, 500*time.Millisecond)
}

func waitForWorkspaceDirectoryMissing(t *testing.T, ctx context.Context, client *apiClient, workspaceID string, machineID string, directoryName string) {
	t.Helper()

	require.Eventually(t, func() bool {
		workspaces := mustDoJSON[apiMachineWorkspacePage](ctx, client, http.MethodGet, fmt.Sprintf("/api/v1/workspaces/%s/machines/%s/workspace-directories", workspaceID, machineID), http.StatusOK, nil)
		return !contains(directoryNames(workspaces.Workspaces), directoryName)
	}, 30*time.Second, 500*time.Millisecond)
}

func requireCommand(t *testing.T, name string) {
	t.Helper()
	_, err := exec.LookPath(name)
	if errors.Is(err, exec.ErrNotFound) {
		t.Skipf("%s is required for e2e", name)
	}
	require.NoError(t, err)
}

func channelIDByName(t *testing.T, channels []apiChannel, name string) string {
	t.Helper()
	for _, channel := range channels {
		if channel.Name == name {
			return channel.ID
		}
	}
	require.FailNowf(t, "channel not found", "name=%s", name)
	return ""
}

func contains(values []string, target string) bool {
	for _, value := range values {
		if value == target {
			return true
		}
	}
	return false
}

func directoryNames(workspaces []apiMachineWorkspace) []string {
	names := make([]string, 0, len(workspaces))
	for _, workspace := range workspaces {
		names = append(names, workspace.DirectoryName)
	}
	return names
}

func filenames(files []workspaceFile) []string {
	names := make([]string, 0, len(files))
	for _, file := range files {
		names = append(names, file.Name)
	}
	return names
}

func stringValue(value *string) string {
	if value == nil {
		return ""
	}
	return *value
}

func parseRunSummaryEntries(t *testing.T, content workspaceFileContent) []fakeCodexRunSummaryEntry {
	t.Helper()

	raw := strings.TrimSpace(stringValue(content.Content))
	require.NotEmpty(t, raw)

	var entries []fakeCodexRunSummaryEntry
	require.NoError(t, json.Unmarshal([]byte(raw), &entries))
	return entries
}

func filterRunSummaryEntriesByReplyTarget(entries []fakeCodexRunSummaryEntry, replyTarget string) []fakeCodexRunSummaryEntry {
	filtered := make([]fakeCodexRunSummaryEntry, 0, len(entries))
	for _, entry := range entries {
		if entry.ReplyTarget != replyTarget {
			continue
		}
		filtered = append(filtered, entry)
	}
	return filtered
}

func threadSummaryKey(channelName string, parentMessageID string) string {
	return fmt.Sprintf("#%s:%s", channelName, strings.ToLower(parentMessageID[:8]))
}