package daemon_test

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/gorilla/websocket"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/api/daemon"
	"github.com/tidbcloud/AgentTeam/internal/api/testkit"
	"github.com/tidbcloud/AgentTeam/internal/archive"
	"github.com/tidbcloud/AgentTeam/internal/core/secrets"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/runtimecontrol"
	"github.com/tidbcloud/AgentTeam/internal/service/agentruntime"
	"github.com/tidbcloud/AgentTeam/internal/store/agent"
	"github.com/tidbcloud/AgentTeam/internal/store/message"
)

// initGinTestMode guards the one-time write of Gin's package-level mode so
// that t.Parallel() tests never race on it. The once completes before any
// parallel test body proceeds (the first caller performs the write; later
// callers block until it finishes), so the shared global is initialized
// exactly once, ahead of the parallel phase.
var initGinTestMode sync.Once

func ensureGinTestMode() {
	initGinTestMode.Do(func() {
		gin.SetMode(gin.TestMode)
	})
}

func seedDaemonTestChannelScope(t *testing.T, pool coredb.QueryExecer, workspaceID string, channelID string, slug string, createdByMemberID string, now time.Time) {
	t.Helper()

	mustExec(t, pool, `
INSERT IGNORE INTO scopes (
    id, workspace_id, type, identity_key, latest_seq, latest_global_seq,
    latest_message_at, archived_until_seq, status, created_by_member_id,
    created_at, updated_at
) VALUES ($1, $2, 'channel', $3, 0, 0, NULL, 0, 'active', NULLIF($4, ''), $5, $5)
`, channelID, workspaceID, "channel:"+channelID, createdByMemberID, now)
	mustExec(t, pool, `
INSERT IGNORE INTO scope_channel_profiles (
    scope_id, workspace_id, slug, display_name, description, visibility,
    owner_member_id, is_all_channel, created_at, updated_at
) VALUES ($1, $2, $3, $3, '', 'public', NULL, CASE WHEN $3 = 'all' THEN 1 ELSE 0 END, $4, $4)
`, channelID, workspaceID, slug, now)
}

func seedDaemonTestScopeMembership(t *testing.T, pool coredb.QueryExecer, scopeID string, memberID string, notificationState string, now time.Time) {
	t.Helper()

	require.NoError(t, message.EnsureScopeMembership(context.Background(), pool, scopeID, memberID, "public", "", notificationState, 0, 0, now))
}

type autoWakeHubFixture struct {
	ctx         context.Context
	dataStore   *testkit.Store
	hub         *daemon.Hub
	conn        *websocket.Conn
	wsURL       string
	pool        coredb.DB
	workspaceID string
	userID      string
	channelID   string
	machineID   string
	agentID     string
	agentHandle string
	ready       <-chan agent.ReadyPayload
}

type readySignalAgentStore struct {
	daemon.AgentStore
	activeTargets daemon.TaskActiveTargetStore
	ready         chan<- agent.ReadyPayload
}

func (store readySignalAgentStore) MarkMachineReady(ctx context.Context, identity agent.MachineIdentity, payload agent.ReadyPayload) error {
	if err := store.AgentStore.MarkMachineReady(ctx, identity, payload); err != nil {
		return err
	}
	store.ready <- payload
	return nil
}

func (store readySignalAgentStore) GetTaskActiveTarget(ctx context.Context, workspaceID string, scope string, targetKey string) (message.ReceiveActiveTaskTarget, bool, error) {
	return store.activeTargets.GetTaskActiveTarget(ctx, workspaceID, scope, targetKey)
}

func TestHubReadyRunningAgentsDeliverBeforeFreshStatus(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-reconnect")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{fixture.agentID},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	sendErrCh := make(chan error, 1)
	go func() {
		sendErrCh <- fixture.notifyAgentMessage(101, "please wake after reconnect")
	}()

	deliver := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:deliver", deliver["type"])
	require.Equal(t, fixture.agentID, deliver["agentId"])

	seq, ok := deliver["seq"].(float64)
	require.True(t, ok)
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:deliver:ack",
		"agentId":   fixture.agentID,
		"messageId": "msg-101",
		"seq":       int(seq),
		"accepted":  true,
	}))

	select {
	case err := <-sendErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for ready-running routed message send")
	}
}

func TestHubReadyLegacyWorkspaceWriteAgentRunsWithoutPermissionModeCapability(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-reconnect-workspace")
	mustExec(t, fixture.pool, `UPDATE agents SET permission_mode = $2 WHERE id = $1`, fixture.agentID, agent.AgentPermissionModeWorkspaceWrite)

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{fixture.agentID},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	sendErrCh := make(chan error, 1)
	go func() {
		sendErrCh <- fixture.notifyAgentMessage(102, "deliver to normalized legacy agent")
	}()

	deliver := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:deliver", deliver["type"])
	require.Equal(t, fixture.agentID, deliver["agentId"])
	seq, ok := deliver["seq"].(float64)
	require.True(t, ok)
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:deliver:ack",
		"agentId":   fixture.agentID,
		"messageId": "msg-102",
		"seq":       int(seq),
		"accepted":  true,
	}))

	select {
	case err := <-sendErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for normalized legacy delivery")
	}
}

func TestHubStartWakeMessageIncludesAgentSessionActiveTaskTarget(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-active-target")
	now := time.Now().UTC()
	mustExec(t, fixture.pool, `UPDATE agents SET runtime = 'claude' WHERE id = $1`, fixture.agentID)
	insertDaemonTaskActiveTargetTaskFixture(t, fixture, "tsk_agent_active_target", 9, "Agent active target", now)
	_, err := fixture.dataStore.PrivateAgents().SetTaskActiveTargetForUser(fixture.ctx, fixture.userID, fixture.workspaceID, message.TaskActiveTargetRecord{
		Scope:            message.TaskActiveTargetScopeAgentSession,
		TargetKey:        fixture.agentID,
		TaskID:           "tsk_agent_active_target",
		TaskChannelID:    fixture.channelID,
		TaskNumber:       9,
		TaskTitle:        "Agent active target",
		Source:           "human_selected",
		ThreadTarget:     "#all",
		AgentID:          fixture.agentID,
		Role:             "worker",
		RootTaskNumber:   9,
		SelectedByUserID: fixture.userID,
	})
	require.NoError(t, err)

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"claude"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.notifyAgentMessage(909, "please use the selected task target"))

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	wakeMessage, ok := start["wakeMessage"].(map[string]any)
	require.True(t, ok)
	activeTarget, ok := wakeMessage["activeTaskTarget"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "#all", activeTarget["channel"])
	require.Equal(t, float64(9), activeTarget["taskNumber"])
	require.Equal(t, "human_selected", activeTarget["source"])
	require.Equal(t, "#all", activeTarget["threadTarget"])
	require.Equal(t, "worker", activeTarget["role"])
}

func TestHubStartWakeMessageSkipsStickyClaimedTaskActiveTarget(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-claimed-target")
	now := time.Now().UTC()
	mustExec(t, fixture.pool, `UPDATE agents SET runtime = 'claude' WHERE id = $1`, fixture.agentID)
	insertDaemonTaskActiveTargetTaskFixture(t, fixture, "tsk_agent_claimed_target", 9, "Agent claimed target", now)
	_, err := fixture.dataStore.PrivateAgents().SetTaskActiveTargetForUser(fixture.ctx, fixture.userID, fixture.workspaceID, message.TaskActiveTargetRecord{
		Scope:            message.TaskActiveTargetScopeAgentSession,
		TargetKey:        fixture.agentID,
		TaskID:           "tsk_agent_claimed_target",
		TaskChannelID:    fixture.channelID,
		TaskNumber:       9,
		TaskTitle:        "Agent claimed target",
		Source:           "claimed_task",
		ThreadTarget:     "#all",
		AgentID:          fixture.agentID,
		Role:             "worker",
		RootTaskNumber:   9,
		SelectedByUserID: fixture.userID,
	})
	require.NoError(t, err)

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"claude"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.notifyAgentMessage(910, "ordinary wake should not inherit a claimed task target"))

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	wakeMessage, ok := start["wakeMessage"].(map[string]any)
	require.True(t, ok)
	require.NotContains(t, wakeMessage, "activeTaskTarget")
}

func TestHubDirectMentionBypassesNoWakePreference(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-direct-mention")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	err := fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
		AgentID:         fixture.agentID,
		MachineID:       fixture.machineID,
		AgentMemberID:   "mem_unused",
		AutoWakeEnabled: false,
		DeliveryReason:  "direct_mention",
	}, 404, message.ReceiveMessage{
		MessageID:   "msg-direct-404",
		ChannelType: "channel",
		ChannelName: "all",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			TargetID:    "owner",
			DisplayName: "owner",
		},
		Content:   "@" + fixture.agentHandle + " please wake despite no-wake default",
		Timestamp: time.Now().UTC(),
	})
	require.NoError(t, err)

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])

	wakeMessage, ok := start["wakeMessage"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "msg-direct-404", wakeMessage["message_id"])
	require.Equal(t, "@"+fixture.agentHandle+" please wake despite no-wake default", wakeMessage["content"])
}

func TestHubTaskAssigneeBypassesNoWakePreference(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-task-assignee")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	notifyErr := make(chan error, 1)
	go func() {
		notifyErr <- fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
			AgentID:         fixture.agentID,
			MachineID:       fixture.machineID,
			AgentMemberID:   "mem_task_assignee",
			AutoWakeEnabled: false,
			DeliveryReason:  message.AgentDeliveryReasonTaskAssignee,
		}, 410, message.ReceiveMessage{
			MessageID:   "activity-task-assignee-410",
			ChannelType: "task",
			ChannelName: "task_scope_410",
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				TargetID:    "owner",
				DisplayName: "owner",
			},
			Content:     "Changed task status to in_review",
			MessageType: "task.status_changed",
			Timestamp:   time.Now().UTC(),
		})
	}()

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
	wakeMessage, ok := start["wakeMessage"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "activity-task-assignee-410", wakeMessage["message_id"])
	require.Equal(t, "Changed task status to in_review", wakeMessage["content"])

	select {
	case err := <-notifyErr:
		t.Fatalf("Task start wake completed before durable daemon acknowledgement: %v", err)
	case <-time.After(50 * time.Millisecond):
	}
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:deliver:ack",
		"agentId":   fixture.agentID,
		"messageId": "activity-task-assignee-410",
		"seq":       0,
		"accepted":  true,
	}))
	select {
	case err := <-notifyErr:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for durably acknowledged Task start wake")
	}
}

func TestHubTaskStartWakeRejectOrMissingAckRemainsRetryable(t *testing.T) {
	// Keep this deadline-sensitive socket handshake ahead of the package's
	// parallel tests so scheduler contention cannot masquerade as a protocol
	// acknowledgement timeout.
	for _, test := range []struct {
		name    string
		sendAck bool
	}{
		{name: "negative_ack", sendAck: true},
		{name: "missing_ack"},
	} {
		t.Run(test.name, func(t *testing.T) {
			fixture := setupAutoWakeHubFixture(t, "agent-task-ack-"+test.name)
			fixture.writeReady(t, map[string]any{
				"type":          "ready",
				"capabilities":  []string{"workspace"},
				"runtimes":      []string{"codex"},
				"runningAgents": []string{},
				"hostname":      "test-host",
				"os":            "linux",
				"daemonVersion": "0.9.0",
			})

			ctx, cancel := context.WithTimeout(fixture.ctx, 250*time.Millisecond)
			defer cancel()
			notifyErr := make(chan error, 1)
			go func() {
				notifyErr <- fixture.hub.NotifyAgentMessage(ctx, message.AgentRecipient{
					AgentID:         fixture.agentID,
					MachineID:       fixture.machineID,
					AgentMemberID:   "mem_task_ack",
					AutoWakeEnabled: true,
					DeliveryReason:  message.AgentDeliveryReasonTaskWatch,
				}, 412, message.ReceiveMessage{
					MessageID:   "activity-task-ack-412",
					ChannelType: "task",
					ChannelName: "task_scope_412",
					AuthorProfile: message.AuthorProfile{
						ActorType:   message.AuthorActorTypeHuman,
						TargetID:    "owner",
						DisplayName: "owner",
					},
					Content:     "Updated task",
					MessageType: "task.updated",
					Timestamp:   time.Now().UTC(),
				})
			}()

			start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
			require.Equal(t, "agent:start", start["type"])
			if test.sendAck {
				require.NoError(t, fixture.conn.WriteJSON(map[string]any{
					"type":      "agent:deliver:ack",
					"agentId":   fixture.agentID,
					"messageId": "activity-task-ack-412",
					"seq":       0,
					"accepted":  false,
					"detail":    "wake was not durably accepted",
				}))
			}

			select {
			case err := <-notifyErr:
				require.Error(t, err)
				if test.sendAck {
					require.ErrorContains(t, err, "wake was not durably accepted")
				} else {
					require.ErrorIs(t, err, context.DeadlineExceeded)
				}
			case <-time.After(2 * time.Second):
				t.Fatal("timeout waiting for failed Task start wake")
			}
		})
	}
}

func TestHubWorkflowStartWakeRequiresDurableDaemonAcknowledgement(t *testing.T) {
	fixture := setupAutoWakeHubFixture(t, "agent-workflow-ack")
	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	notifyErr := make(chan error, 1)
	go func() {
		notifyErr <- fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
			AgentID:         fixture.agentID,
			MachineID:       fixture.machineID,
			AgentMemberID:   "mem_workflow_ack",
			AutoWakeEnabled: true,
			DeliveryReason:  message.AgentDeliveryReasonDirectMention,
		}, 413, message.ReceiveMessage{
			MessageID:   "msg-workflow-ack-413",
			ChannelType: message.WorkflowScopeType,
			ChannelName: "pln_workflow_ack",
			AuthorProfile: message.AuthorProfile{
				ActorType:   message.AuthorActorTypeHuman,
				TargetID:    "owner",
				DisplayName: "owner",
			},
			Content:     "Update the Workflow draft.",
			MessageType: "text",
			Timestamp:   time.Now().UTC(),
		})
	}()

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	wakeMessage, ok := start["wakeMessage"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "msg-workflow-ack-413", wakeMessage["message_id"])
	select {
	case err := <-notifyErr:
		t.Fatalf("Workflow start wake completed before durable daemon acknowledgement: %v", err)
	default:
	}

	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:deliver:ack",
		"agentId":   fixture.agentID,
		"messageId": "msg-workflow-ack-413",
		"seq":       0,
		"accepted":  true,
	}))
	select {
	case err := <-notifyErr:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for durably acknowledged Workflow start wake")
	}
}

func TestHubUnwatchedTaskRecipientDoesNotWake(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-task-unwatched")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	err := fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
		AgentID:         fixture.agentID,
		MachineID:       fixture.machineID,
		AgentMemberID:   "mem_task_unwatched",
		AutoWakeEnabled: false,
		DeliveryReason:  message.AgentDeliveryReasonTaskWatch,
	}, 411, message.ReceiveMessage{
		MessageID:   "activity-task-unwatched-411",
		ChannelType: "task",
		ChannelName: "task_scope_411",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			TargetID:    "owner",
			DisplayName: "owner",
		},
		Content:     "Updated task",
		MessageType: "task.updated",
		Timestamp:   time.Now().UTC(),
	})
	require.NoError(t, err)

	require.NoError(t, fixture.conn.SetReadDeadline(time.Now().Add(250*time.Millisecond)))
	_, _, readErr := fixture.conn.ReadMessage()
	require.Error(t, readErr)
	require.True(t, isTimeout(readErr), "expected no task wake for an unwatched Task recipient, got: %v", readErr)
}

func TestHubTeamMentionBypassesNoWakePreference(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-team-mention")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	recipient := message.AgentRecipient{
		AgentID:         fixture.agentID,
		MachineID:       fixture.machineID,
		AgentMemberID:   "mem_unused",
		AutoWakeEnabled: false,
		DeliveryReason:  message.AgentDeliveryReasonTeamMention,
		TeamContexts: []message.TeamDeliveryContext{
			{
				TeamID:          "team_testteam",
				TeamHandle:      "testteam",
				TeamDisplayName: "testteam",
				TeamMemberID:    "tm_mem_test",
				RoleLabel:       "dev",
			},
		},
	}
	receiveMessage := message.ReceiveMessageWithRecipientTeamContexts(message.ReceiveMessage{
		MessageID:   "msg-team-409",
		ChannelType: "channel",
		ChannelName: "testtest",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			TargetID:    "owner",
			DisplayName: "owner",
		},
		Content:   "@testteam please wake despite no-wake default",
		Timestamp: time.Now().UTC(),
	}, recipient)
	err := fixture.hub.NotifyAgentMessage(fixture.ctx, recipient, 409, receiveMessage)
	require.NoError(t, err)

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])

	wakeMessage, ok := start["wakeMessage"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "msg-team-409", wakeMessage["message_id"])
	require.Equal(t, "@testteam please wake despite no-wake default", wakeMessage["content"])
	teamContexts, ok := wakeMessage["teamContexts"].([]any)
	require.True(t, ok)
	require.Len(t, teamContexts, 1)
	teamContext, ok := teamContexts[0].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "testteam", teamContext["teamHandle"])
	require.Equal(t, "dev", teamContext["roleLabel"])
}

func TestHubTaskCommandNotifyBypassesNoWakePreference(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-task-command-notify")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	err := fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
		AgentID:         fixture.agentID,
		MachineID:       fixture.machineID,
		AgentMemberID:   "mem_unused",
		AutoWakeEnabled: false,
		DeliveryReason:  message.AgentDeliveryReasonTaskCommandNotify,
	}, 406, message.ReceiveMessage{
		MessageID:   "msg-task-command-406",
		MessageType: message.MessageTypeTaskDispatch,
		ChannelType: "channel",
		ChannelName: "all",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			TargetID:    "owner",
			DisplayName: "owner",
		},
		Content:   "Task discussion update for #t13",
		Timestamp: time.Now().UTC(),
	})
	require.NoError(t, err)

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])

	wakeMessage, ok := start["wakeMessage"].(map[string]any)
	require.True(t, ok)
	require.Equal(t, "msg-task-command-406", wakeMessage["message_id"])
	require.Equal(t, message.MessageTypeTaskDispatch, wakeMessage["message_type"])
	require.Equal(t, "Task discussion update for #t13", wakeMessage["content"])
}

func TestHubControlledBroadcastBypassesNoWakePreference(t *testing.T) {
	t.Parallel()
	tests := []struct {
		name      string
		reason    string
		messageID string
		content   string
	}{
		{
			name:      "all",
			reason:    message.AgentDeliveryReasonChannelBroadcastAll,
			messageID: "msg-broadcast-all-407",
			content:   "@all please wake every channel agent",
		},
		{
			name:      "online",
			reason:    message.AgentDeliveryReasonChannelBroadcastOnline,
			messageID: "msg-broadcast-online-408",
			content:   "@online please wake connected channel agents",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			fixture := setupAutoWakeHubFixture(t, "agent-broadcast-"+test.name)

			fixture.writeReady(t, map[string]any{
				"type":          "ready",
				"capabilities":  []string{"workspace"},
				"runtimes":      []string{"codex"},
				"runningAgents": []string{},
				"hostname":      "test-host",
				"os":            "linux",
				"daemonVersion": "0.9.0",
			})

			err := fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
				AgentID:         fixture.agentID,
				MachineID:       fixture.machineID,
				AgentMemberID:   "mem_unused",
				AutoWakeEnabled: false,
				DeliveryReason:  test.reason,
			}, 407, message.ReceiveMessage{
				MessageID:   test.messageID,
				ChannelType: "channel",
				ChannelName: "all",
				AuthorProfile: message.AuthorProfile{
					ActorType:   message.AuthorActorTypeHuman,
					TargetID:    "owner",
					DisplayName: "owner",
				},
				Content:   test.content,
				Timestamp: time.Now().UTC(),
			})
			require.NoError(t, err)

			start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
			require.Equal(t, "agent:start", start["type"])
			require.Equal(t, fixture.agentID, start["agentId"])

			wakeMessage, ok := start["wakeMessage"].(map[string]any)
			require.True(t, ok)
			require.Equal(t, test.messageID, wakeMessage["message_id"])
			require.Equal(t, test.content, wakeMessage["content"])
		})
	}
}

func TestHubDirectMentionDoesNotAutoWakeManuallyStoppedAgent(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-manually-stopped")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.hub.StopAgent(fixture.machineID, fixture.agentID))
	stop := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:stop", stop["type"])
	require.Equal(t, fixture.agentID, stop["agentId"])

	err := fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
		AgentID:         fixture.agentID,
		MachineID:       fixture.machineID,
		AgentMemberID:   "mem_unused",
		AutoWakeEnabled: false,
		DeliveryReason:  "direct_mention",
	}, 405, message.ReceiveMessage{
		MessageID:   "msg-direct-405",
		ChannelType: "channel",
		ChannelName: "all",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			TargetID:    "owner",
			DisplayName: "owner",
		},
		Content:   "@" + fixture.agentHandle + " stay queued until manual start",
		Timestamp: time.Now().UTC(),
	})
	require.NoError(t, err)

	_, _, err = fixture.conn.ReadMessage()
	require.True(t, isTimeout(err), "expected no agent:start while agent is manually stopped, got: %v", err)
}

func TestHubWorkflowWakeDoesNotRestartManuallyStoppedAgent(t *testing.T) {
	fixture := setupAutoWakeHubFixture(t, "agent-workflow-manually-stopped")
	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.hub.StopAgent(fixture.machineID, fixture.agentID))
	stop := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:stop", stop["type"])

	err := fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
		AgentID:         fixture.agentID,
		MachineID:       fixture.machineID,
		AgentMemberID:   "mem_workflow_stopped",
		AutoWakeEnabled: true,
		DeliveryReason:  message.AgentDeliveryReasonDirectMention,
	}, 414, message.ReceiveMessage{
		MessageID:   "msg-workflow-stopped-414",
		ChannelType: message.WorkflowScopeType,
		ChannelName: "pln_workflow_stopped",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			TargetID:    "owner",
			DisplayName: "owner",
		},
		Content:     "Do not restart until I explicitly start you.",
		MessageType: "text",
		Timestamp:   time.Now().UTC(),
	})
	require.ErrorIs(t, err, runtimecontrol.ErrAgentManuallyStopped)

	_, _, err = fixture.conn.ReadMessage()
	require.True(t, isTimeout(err), "expected no Workflow agent:start while manually stopped, got: %v", err)
}

func TestHubResetAgentSessionKeepsAutoWakeEligible(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-reset-session")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.hub.ResetAgentSession(fixture.machineID, fixture.agentID))
	stop := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:stop", stop["type"])
	require.Equal(t, fixture.agentID, stop["agentId"])
	require.Equal(t, true, stop["clearSession"])

	require.NoError(t, fixture.notifyAgentMessage(406, "wake after session reset"))
	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
}

func TestHubInterruptAgentRunPreservesSession(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-interrupt-run")

	fixture.writeReady(t, map[string]any{
		"type":                 "ready",
		"capabilities":         []string{"workspace"},
		"runtimes":             []string{"codex"},
		"runningAgents":        []string{fixture.agentID},
		"runningAgentSessions": map[string]string{fixture.agentID: "codex-thread-1"},
		"hostname":             "test-host",
		"os":                   "linux",
		"daemonVersion":        "0.9.0",
	})

	require.NoError(t, fixture.hub.InterruptAgentRun(fixture.machineID, fixture.agentID))
	interrupt := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:interrupt", interrupt["type"])
	require.Equal(t, fixture.agentID, interrupt["agentId"])
	require.NotContains(t, interrupt, "clearSession")

	snapshot := fixture.hub.WorkspaceRuntimeSnapshot(fixture.workspaceID)
	require.Len(t, snapshot.Machines, 1)
	require.Equal(t, fixture.machineID, snapshot.Machines[0].MachineID)
	require.Len(t, snapshot.Machines[0].Agents, 1)
	require.Equal(t, fixture.agentID, snapshot.Machines[0].Agents[0].AgentID)
	require.Equal(t, "codex-thread-1", snapshot.Machines[0].Agents[0].SessionID)
}

func TestHubRejectedRunningAgentDeliveryFallsBackToStart(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-fallback")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})
	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID))

	initialStart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", initialStart["type"])
	require.Equal(t, fixture.agentID, initialStart["agentId"])

	sendErrCh := make(chan error, 1)
	go func() {
		sendErrCh <- fixture.notifyAgentMessage(202, "please fall back to fresh start")
	}()

	deliver := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:deliver", deliver["type"])
	require.Equal(t, fixture.agentID, deliver["agentId"])

	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:deliver:ack",
		"agentId":   fixture.agentID,
		"messageId": "msg-202",
		"seq":       int(deliver["seq"].(float64)),
		"accepted":  false,
		"detail":    "agent is not running on daemon",
	}))

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])

	select {
	case err := <-sendErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for routed message send to finish after fallback")
	}
}

func TestHubInvalidFallbackStartFailureClearsRunningAndStarting(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-invalid-fallback-resolution")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})
	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID))

	initialStart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", initialStart["type"])
	require.Equal(t, fixture.agentID, initialStart["agentId"])

	sendErrCh := make(chan error, 1)
	go func() {
		sendErrCh <- fixture.notifyAgentMessage(203, "fall back to an invalid start wake")
	}()

	deliver := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:deliver", deliver["type"])
	require.Equal(t, fixture.agentID, deliver["agentId"])
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:deliver:ack",
		"agentId":   fixture.agentID,
		"messageId": "msg-203",
		"seq":       int(deliver["seq"].(float64)),
		"accepted":  false,
		"detail":    "unsupported interruptPolicy",
	}))

	fallbackStart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", fallbackStart["type"])
	require.Equal(t, fixture.agentID, fallbackStart["agentId"])
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID),
		"fallback start owns both the running and starting projections until Desktop resolves it")

	select {
	case err := <-sendErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for invalid fallback start command")
	}

	// These are the authoritative terminal frames emitted when Desktop rejects
	// the fallback start wake before launching a Process. The seq-zero negative
	// delivery ACK is intentionally not part of the Server start lifecycle.
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:wake_status",
		"agentId":   fixture.agentID,
		"messageId": "msg-203",
		"wakeToken": "agent-start:msg-203",
		"target":    "#all",
		"status":    agent.AgentRuntimeEventWakeStatusFailed,
		"detail":    "invalid agent:start wake: unsupported interruptPolicy",
		"timestamp": time.Now().UTC().Format(time.RFC3339Nano),
	}))
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":       "agent:runtime_event",
		"agentId":    fixture.agentID,
		"messageId":  "msg-203",
		"eventClass": agent.AgentRuntimeEventClassRuntimeStartFailed,
		"target":     "#all",
		"detail":     "invalid agent:start wake: unsupported interruptPolicy",
		"timestamp":  time.Now().UTC().Format(time.RFC3339Nano),
	}))
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":    "agent:status",
		"agentId": fixture.agentID,
		"status":  "inactive",
	}))

	require.Eventually(t, func() bool {
		return !fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID)
	}, 2*time.Second, 20*time.Millisecond,
		"runtime-start failure must clear both fallback running and starting state")

	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID),
		"a terminally rejected fallback must not block a later explicit start")
	retryStart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", retryStart["type"])
	require.Equal(t, fixture.agentID, retryStart["agentId"])
}

func TestHubRuntimeFailureEventClearsRunningAfterAcceptedDeliver(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-runtime-failure-clears-running")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})
	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID))

	initialStart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", initialStart["type"])
	require.Equal(t, fixture.agentID, initialStart["agentId"])

	sendErrCh := make(chan error, 1)
	go func() {
		sendErrCh <- fixture.notifyAgentMessage(204, "this accepted delivery fails locally")
	}()

	deliver := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:deliver", deliver["type"])
	require.Equal(t, fixture.agentID, deliver["agentId"])

	seq, ok := deliver["seq"].(float64)
	require.True(t, ok)
	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":      "agent:deliver:ack",
		"agentId":   fixture.agentID,
		"messageId": "msg-204",
		"seq":       int(seq),
		"accepted":  true,
	}))

	select {
	case err := <-sendErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for accepted routed message send to finish")
	}

	require.NoError(t, fixture.conn.WriteJSON(map[string]any{
		"type":       "agent:runtime_event",
		"agentId":    fixture.agentID,
		"messageId":  "msg-204",
		"eventClass": agent.AgentRuntimeEventClassWakeFailed,
		"target":     "#all",
		"detail":     "init MEMORY.md: MEMORY.md exceeds 32768 characters",
		"timestamp":  time.Now().UTC().Format(time.RFC3339Nano),
	}))

	var startErr error
	require.Eventually(t, func() bool {
		startErr = fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID)
		return startErr == nil
	}, 2*time.Second, 20*time.Millisecond, "last start error: %v", startErr)

	restart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", restart["type"])
	require.Equal(t, fixture.agentID, restart["agentId"])
}

func TestHubExplicitStartRejectsWhileStopIsStillPending(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-stop-pending")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})
	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID))

	initialStart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", initialStart["type"])
	require.Equal(t, fixture.agentID, initialStart["agentId"])

	require.NoError(t, fixture.hub.StopAgent(fixture.machineID, fixture.agentID))

	stop := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:stop", stop["type"])
	require.Equal(t, fixture.agentID, stop["agentId"])

	err := fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID)
	require.Error(t, err)
	require.ErrorContains(t, err, "still running")

	require.NoError(t, fixture.conn.SetReadDeadline(time.Now().Add(250*time.Millisecond)))
	_, _, readErr := fixture.conn.ReadMessage()
	require.Error(t, readErr)
	require.True(t, isTimeout(readErr), "expected read timeout, got: %v", readErr)
}

func TestHubRestartAgentReplacesRunningAgent(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-restart-config")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{fixture.agentID},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.hub.RestartAgent(fixture.ctx, fixture.machineID, fixture.agentID))
	restart := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", restart["type"])
	require.Equal(t, fixture.agentID, restart["agentId"])
}

func TestHubStartMarksAgentRunningBeforePreparingStart(t *testing.T) {
	t.Parallel()
	prepareStarted := make(chan struct{})
	releasePrepare := make(chan struct{})
	fixture := setupAutoWakeHubFixtureWithPrepareStart(t, "agent-starting-before-prepare", func(runtimeStore *agentruntime.Service) daemon.PrepareAgentStartFunc {
		return func(ctx context.Context, agentID string, publicBaseURL string, wakeMessage *message.ReceiveMessage, allowWorkspaceWrite bool, includeLegacySession bool, useDirectMaaS bool) (agent.StartCommand, error) {
			close(prepareStarted)
			select {
			case <-releasePrepare:
			case <-ctx.Done():
				return agent.StartCommand{}, ctx.Err()
			}
			return runtimeStore.PrepareAgentStart(ctx, agentID, publicBaseURL, wakeMessage, allowWorkspaceWrite, includeLegacySession, useDirectMaaS)
		}
	})

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	startErrCh := make(chan error, 1)
	go func() {
		startErrCh <- fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID)
	}()

	select {
	case <-prepareStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for PrepareAgentStart")
	}
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.1",
	})
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	close(releasePrepare)
	select {
	case err := <-startErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for StartAgent")
	}

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
}

func TestHubReadyCapabilityDowngradeKeepsLegacyWorkspaceWriteStart(t *testing.T) {
	t.Parallel()
	prepareStarted := make(chan struct{})
	releasePrepare := make(chan struct{})
	fixture := setupAutoWakeHubFixtureWithPrepareStart(t, "agent-starting-cap-downgrade", func(runtimeStore *agentruntime.Service) daemon.PrepareAgentStartFunc {
		return func(ctx context.Context, agentID string, publicBaseURL string, wakeMessage *message.ReceiveMessage, allowWorkspaceWrite bool, includeLegacySession bool, useDirectMaaS bool) (agent.StartCommand, error) {
			close(prepareStarted)
			select {
			case <-releasePrepare:
			case <-ctx.Done():
				return agent.StartCommand{}, ctx.Err()
			}
			return runtimeStore.PrepareAgentStart(ctx, agentID, publicBaseURL, wakeMessage, allowWorkspaceWrite, includeLegacySession, useDirectMaaS)
		}
	})
	mustExec(t, fixture.pool, `UPDATE agents SET permission_mode = $2 WHERE id = $1`, fixture.agentID, agent.AgentPermissionModeWorkspaceWrite)

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace", "agent:permission-mode"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	startErrCh := make(chan error, 1)
	go func() {
		startErrCh <- fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID)
	}()

	select {
	case <-prepareStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for PrepareAgentStart")
	}
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.1",
	})

	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	close(releasePrepare)
	select {
	case err := <-startErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for StartAgent")
	}

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
}

func TestHubReadyCapabilityDowngradeKeepsLegacyWorkspaceWriteRestart(t *testing.T) {
	t.Parallel()
	prepareStarted := make(chan struct{})
	releasePrepare := make(chan struct{})
	fixture := setupAutoWakeHubFixtureWithPrepareStart(t, "agent-restart-cap-downgrade", func(runtimeStore *agentruntime.Service) daemon.PrepareAgentStartFunc {
		return func(ctx context.Context, agentID string, publicBaseURL string, wakeMessage *message.ReceiveMessage, allowWorkspaceWrite bool, includeLegacySession bool, useDirectMaaS bool) (agent.StartCommand, error) {
			close(prepareStarted)
			select {
			case <-releasePrepare:
			case <-ctx.Done():
				return agent.StartCommand{}, ctx.Err()
			}
			return runtimeStore.PrepareAgentStart(ctx, agentID, publicBaseURL, wakeMessage, allowWorkspaceWrite, includeLegacySession, useDirectMaaS)
		}
	})
	mustExec(t, fixture.pool, `UPDATE agents SET permission_mode = $2 WHERE id = $1`, fixture.agentID, agent.AgentPermissionModeWorkspaceWrite)

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace", "agent:permission-mode"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{fixture.agentID},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	restartErrCh := make(chan error, 1)
	go func() {
		restartErrCh <- fixture.hub.RestartAgent(fixture.ctx, fixture.machineID, fixture.agentID)
	}()

	select {
	case <-prepareStarted:
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for PrepareAgentStart")
	}
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.1",
	})

	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))

	close(releasePrepare)
	select {
	case err := <-restartErrCh:
		require.NoError(t, err)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for RestartAgent")
	}

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))
}

func TestHubRestartLegacyWorkspaceWriteAgentDoesNotRequirePermissionModeCapability(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-restart-permission-cap")
	oldToken := "workspace-write-token-before-downgrade"
	mustExec(t, fixture.pool, `
UPDATE agents
SET permission_mode = $2,
    auth_token_hash = $3
WHERE id = $1
`, fixture.agentID, agent.AgentPermissionModeWorkspaceWrite, secrets.HashToken(oldToken))

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{fixture.agentID},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})
	require.True(t, fixture.hub.AgentRuntimeRunning(fixture.machineID, fixture.agentID))
	_, err := agent.New(fixture.pool).AuthenticateAgent(fixture.ctx, fixture.agentID, oldToken)
	require.NoError(t, err)

	err = fixture.hub.RestartAgent(fixture.ctx, fixture.machineID, fixture.agentID)
	require.NoError(t, err)

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
	config, ok := start["config"].(map[string]any)
	require.True(t, ok, "expected config object in %#v", start)
	require.Equal(t, agent.AgentPermissionModeFullAccess, config["permissionMode"])
}

func TestHubReadyWithoutPermissionCapabilityKeepsLegacyWorkspaceWriteToken(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-ready-revoke-unreported")
	oldToken := "workspace-write-token-unreported"
	mustExec(t, fixture.pool, `
UPDATE agents
SET permission_mode = $2,
    auth_token_hash = $3
WHERE id = $1
`, fixture.agentID, agent.AgentPermissionModeWorkspaceWrite, secrets.HashToken(oldToken))

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	_, err := agent.New(fixture.pool).AuthenticateAgent(fixture.ctx, fixture.agentID, oldToken)
	require.NoError(t, err)

	require.NoError(t, fixture.conn.SetReadDeadline(time.Now().Add(250*time.Millisecond)))
	_, _, readErr := fixture.conn.ReadMessage()
	require.Error(t, readErr)
	require.True(t, isTimeout(readErr), "expected read timeout, got: %v", readErr)
}

func TestStartCodexAgentDoesNotRequireLegacyCapability(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-codex-manual")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID))

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
}

func TestStartLegacyWorkspaceWriteAgentDoesNotRequirePermissionModeCapability(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-workspace-permission-cap")
	mustExec(t, fixture.pool, `UPDATE agents SET permission_mode = $2 WHERE id = $1`, fixture.agentID, agent.AgentPermissionModeWorkspaceWrite)

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID))

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
	config, ok := start["config"].(map[string]any)
	require.True(t, ok, "expected config object in %#v", start)
	require.Equal(t, agent.AgentPermissionModeFullAccess, config["permissionMode"])
}

func TestStartLegacyWorkspaceWriteAgentWithPermissionModeCapabilityNormalizesToFullAccess(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-workspace-permission-ok")
	mustExec(t, fixture.pool, `UPDATE agents SET permission_mode = $2 WHERE id = $1`, fixture.agentID, agent.AgentPermissionModeWorkspaceWrite)

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace", "agent:permission-mode"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	require.NoError(t, fixture.hub.StartAgent(fixture.ctx, fixture.machineID, fixture.agentID))

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
	config, ok := start["config"].(map[string]any)
	require.True(t, ok, "expected config object in %#v", start)
	require.Equal(t, agent.AgentPermissionModeFullAccess, config["permissionMode"])
}

func TestAutoWakeCodexAgentDoesNotRequireLegacyCapability(t *testing.T) {
	t.Parallel()
	fixture := setupAutoWakeHubFixture(t, "agent-codex-autowake")

	fixture.writeReady(t, map[string]any{
		"type":          "ready",
		"capabilities":  []string{"workspace"},
		"runtimes":      []string{"codex"},
		"runningAgents": []string{},
		"hostname":      "test-host",
		"os":            "linux",
		"daemonVersion": "0.9.0",
	})

	messageID := int64(303)
	require.NoError(t, fixture.notifyAgentMessage(messageID, "please wake without legacy capability"))

	start := readWSJSONMessage(t, fixture.conn, 2*time.Second)
	require.Equal(t, "agent:start", start["type"])
	require.Equal(t, fixture.agentID, start["agentId"])
}

func setupAutoWakeHubFixture(t *testing.T, agentHandle string) autoWakeHubFixture {
	return setupAutoWakeHubFixtureWithPrepareStart(t, agentHandle, nil)
}

func setupAutoWakeHubFixtureWithPrepareStart(t *testing.T, agentHandle string, prepareStart func(*agentruntime.Service) daemon.PrepareAgentStartFunc) autoWakeHubFixture {
	t.Helper()

	ctx := context.Background()
	pool := openDaemonIntegrationTestTiDB(t)

	archiveStorage, err := archive.NewFilesystem(t.TempDir())
	require.NoError(t, err)
	dataStore := testkit.New(pool, archiveStorage, 30)

	ensureGinTestMode()
	logger := slog.New(slog.NewTextHandler(io.Discard, nil))

	server := httptest.NewUnstartedServer(http.NotFoundHandler())
	baseURL := "http://" + server.Listener.Addr().String()
	runtimeStore := agentruntime.New(agentruntime.Dependencies{Agents: dataStore.Agent(), Messages: dataStore.Message(), Recovery: dataStore.Rewind()})
	prepareAgentStart := runtimeStore.PrepareAgentStart
	if prepareStart != nil {
		prepareAgentStart = prepareStart(runtimeStore)
	}
	ready := make(chan agent.ReadyPayload, 1)
	agents := readySignalAgentStore{AgentStore: runtimeStore.Agents(), activeTargets: dataStore.Agent(), ready: ready}
	hub := daemon.New(logger, daemon.NewStore(agents, runtimeStore.Recovery(), prepareAgentStart), baseURL, nil, dataStore.AgentOps())
	router := gin.New()
	daemon.RegisterRoutes(router, hub, runtimeStore.Agents(), nil, daemon.RegisterRoutesOptions{})
	server.Config.Handler = router
	server.Start()
	t.Cleanup(server.Close)

	now := time.Now().UTC()
	workspaceID := "wsp_" + agentHandle
	userID := "usr_" + agentHandle
	humanMemberID := "mem_human_" + agentHandle
	agentMemberID := "mem_agent_" + agentHandle
	channelID := "chn_all_" + agentHandle + agentHandle
	machineID := "mac_" + agentHandle
	machineAPIKey := "machine-key-" + agentHandle
	agentID := "agt_" + agentHandle

	mustExec(t, pool, `
	INSERT INTO users (id, email, name, display_name, avatar_url, is_admin, password_hash, email_verified_at, created_at, updated_at)
	VALUES ($1, $2, $3, $3, '', TRUE, 'hash', $4, $4, $4)
	`, userID, agentHandle+"@example.com", agentHandle+"-owner", now)
	organizationID := seedDaemonExternalTestWorkspaceOrganization(t, pool, workspaceID, agentHandle, userID, now)
	mustExec(t, pool, `
	INSERT INTO workspaces (id, organization_id, slug, name, latest_message_seq, created_at, updated_at)
	VALUES ($1, $2, $3, $3, 1, $4, $4)
	`, workspaceID, organizationID, agentHandle, now)
	mustExec(t, pool, `
INSERT INTO members (id, workspace_id, type, display_name, user_id, workspace_role, created_at, updated_at)
VALUES ($1, $2, 'human', 'Human', $3, 'owner', $4, $4)
`, humanMemberID, workspaceID, userID, now)
	mustExec(t, pool, `
INSERT INTO members (id, workspace_id, type, display_name, created_at, updated_at)
VALUES ($1, $2, 'agent', 'App', $3, $3)
`, agentMemberID, workspaceID, now)
	seedDaemonTestChannelScope(t, pool, workspaceID, channelID, "all", humanMemberID, now)
	seedDaemonTestScopeMembership(t, pool, channelID, humanMemberID, "active", now)
	seedDaemonTestScopeMembership(t, pool, channelID, agentMemberID, "active", now)
	mustExec(t, pool, `
INSERT INTO machines (id, workspace_id, created_by_user_id, name, api_key, created_at, updated_at)
VALUES ($1, $2, $3, 'Test machine', $4, $5, $5)
`, machineID, workspaceID, userID, machineAPIKey, now)
	mustExec(t, pool, `
INSERT INTO agents (
	id, workspace_id, machine_id, member_id, name, runtime, model, description,
	auth_token_hash, auto_wake_enabled, status, activity, activity_detail, created_at, updated_at
) VALUES (
	$1, $2, $3, $4, $5, 'codex', 'gpt-5-codex', '',
	$6, TRUE, 'active', 'online', '', $7, $7
)
`, agentID, workspaceID, machineID, agentMemberID, agentHandle, secrets.HashToken("placeholder"), now)
	seedDaemonTestScopeMembership(t, pool, channelID, agentMemberID, "auto-wake", now)

	wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/api/v1/workspaces/" + workspaceID + "/daemon/connect?key=" + machineAPIKey
	conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil)
	require.NoError(t, err)
	t.Cleanup(func() { _ = conn.Close() })
	requireServerCapabilitiesFrame(t, conn)

	return autoWakeHubFixture{
		ctx:         ctx,
		dataStore:   dataStore,
		hub:         hub,
		conn:        conn,
		wsURL:       wsURL,
		pool:        pool,
		workspaceID: workspaceID,
		userID:      userID,
		channelID:   channelID,
		machineID:   machineID,
		agentID:     agentID,
		agentHandle: agentHandle,
		ready:       ready,
	}
}

func (fixture autoWakeHubFixture) connect(t *testing.T) *websocket.Conn {
	t.Helper()

	conn, _, err := websocket.DefaultDialer.Dial(fixture.wsURL, nil)
	require.NoError(t, err)
	t.Cleanup(func() { _ = conn.Close() })
	requireServerCapabilitiesFrame(t, conn)
	return conn
}

func requireServerCapabilitiesFrame(t *testing.T, conn *websocket.Conn) {
	t.Helper()

	frame := readWSJSONMessage(t, conn, 2*time.Second)
	require.Equal(t, "server:capabilities", frame["type"])
	capabilities, ok := frame["capabilities"].([]any)
	require.True(t, ok)
	require.Contains(t, capabilities, "transient_legacy_wake_status")
}

func (fixture autoWakeHubFixture) writeReady(t *testing.T, payload map[string]any) {
	t.Helper()

	fixture.writeReadyOn(t, fixture.conn, payload)
}

func (fixture autoWakeHubFixture) writeReadyOn(t *testing.T, conn *websocket.Conn, payload map[string]any) {
	t.Helper()

	require.NoError(t, conn.WriteJSON(payload))

	expectedVersion, _ := payload["daemonVersion"].(string)
	select {
	case ready := <-fixture.ready:
		require.Equal(t, expectedVersion, ready.DaemonVersion)
	case <-time.After(2 * time.Second):
		t.Fatal("timeout waiting for daemon ready processing")
	}
}

func (fixture autoWakeHubFixture) notifyAgentMessage(seq int64, body string) error {
	if fixture.hub == nil {
		return errors.New("hub fixture is not initialized")
	}
	return fixture.hub.NotifyAgentMessage(fixture.ctx, message.AgentRecipient{
		AgentID:         fixture.agentID,
		MachineID:       fixture.machineID,
		AgentMemberID:   "mem_unused",
		AutoWakeEnabled: true,
		DeliveryReason:  "mention",
	}, seq, message.ReceiveMessage{
		MessageID:   fmt.Sprintf("msg-%d", seq),
		ChannelType: "channel",
		ChannelName: "all",
		AuthorProfile: message.AuthorProfile{
			ActorType:   message.AuthorActorTypeHuman,
			TargetID:    "owner",
			DisplayName: "owner",
		},
		Content:   "@" + fixture.agentHandle + " " + body,
		Timestamp: time.Now().UTC(),
	})
}

func readWSJSONMessage(t *testing.T, conn *websocket.Conn, timeout time.Duration) map[string]any {
	t.Helper()

	require.NoError(t, conn.SetReadDeadline(time.Now().Add(timeout)))
	_, data, err := conn.ReadMessage()
	require.NoError(t, err)

	var message map[string]any
	require.NoError(t, json.Unmarshal(data, &message))
	return message
}

func mustExec(t *testing.T, pool coredb.QueryExecer, query string, args ...any) {
	t.Helper()
	_, err := pool.Exec(context.Background(), query, args...)
	require.NoError(t, err)
}

func seedDaemonExternalTestWorkspaceOrganization(t *testing.T, pool coredb.QueryExecer, workspaceID string, slug string, createdByUserID string, now time.Time) string {
	t.Helper()

	organizationID := "org_" + workspaceID
	mustExec(t, pool, `
INSERT INTO organizations (id, slug, name, created_by_user_id, created_at, updated_at)
VALUES ($1, $2, $2, $3, $4, $4)
`, organizationID, slug, createdByUserID, now)
	return organizationID
}

func insertDaemonTaskActiveTargetTaskFixture(t *testing.T, fixture autoWakeHubFixture, taskID string, taskNumber int64, title string, now time.Time) {
	t.Helper()
	boardID := "tbd_" + taskID
	ownerMemberID := "mem_human_" + fixture.agentHandle
	mustExec(t, fixture.pool, `
INSERT INTO task_boards (id, channel_id, workspace_id, next_task_number, created_at, updated_at)
VALUES ($1, $2, $3, 100, $4, $4)
`, boardID, fixture.channelID, fixture.workspaceID, now)
	mustExec(t, fixture.pool, `
INSERT INTO scopes (id, workspace_id, type, identity_key, status, created_by_member_id, created_at, updated_at)
VALUES ($1, $2, 'task', CONCAT('scope/task/', $1), 'active', $3, $4, $4)
`, taskID, fixture.workspaceID, ownerMemberID, now)
	mustExec(t, fixture.pool, `
INSERT INTO tasks (
  id, task_scope_id, board_id, channel_id, task_number, title, status, created_by_member_id,
  created_at, updated_at
) VALUES ($1, $1, $2, $3, $4, $5, 'open', $6, $7, $7)
`, taskID, boardID, fixture.channelID, taskNumber, title, ownerMemberID, now)
}

func isTimeout(err error) bool {
	var netErr net.Error
	return errors.As(err, &netErr) && netErr.Timeout()
}