package server

// P0 #1 production-path coverage: the formal server/Wire worker tick must scan
// awaiting send_message.callback effects and run the automatic RecoveryRunner
// through the corrected canonical working-state judgment (P0 #2), while the
// lease Reconciler and queue-lease behavior stay unchanged.
//
// These tests build the exact production worker via newWorkflowRunWorkerForConfig
// (the wire.go production composition) over a real migrated TiDB database and
// drive one real worker pass (Worker.RunOnce — the same function the server
// ticker calls every second). Only the message/approval adapters are fakes; the
// store, the recovery runner, the reducer, the effect dispatcher, and the lease
// reconciler are all the real production components.

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"testing"
	"time"

	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/AgentTeam/internal/config"
	"github.com/tidbcloud/AgentTeam/internal/coredb"
	"github.com/tidbcloud/AgentTeam/internal/coredb/testkit"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/plan"
	"github.com/tidbcloud/AgentTeam/internal/store/workflow/run"
	invocationlang "github.com/tidbcloud/AgentTeam/internal/workflow/invocation"
	"github.com/tidbcloud/AgentTeam/internal/workflow/orchestrator"
)

// prodRecordingWriter records recovery nudges written by the production
// RecoveryRunner, mirroring the message-writer boundary (WriteRunMessage) used
// by the production wire.
type prodRecordingWriter struct {
	writes []orchestrator.MessageRequest
}

func (w *prodRecordingWriter) WriteRunMessage(_ context.Context, request orchestrator.MessageRequest) (orchestrator.MessageResult, error) {
	w.writes = append(w.writes, request)
	return orchestrator.MessageResult{MessageID: "msg_prod_rec", DeliveryScope: request.Scope}, nil
}

type prodStubApprovalWriter struct{}

func (prodStubApprovalWriter) WriteApproval(context.Context, orchestrator.ApprovalRequest) (orchestrator.ApprovalReceipt, error) {
	return orchestrator.ApprovalReceipt{}, nil
}

type prodStubAssigneeResolver struct{}

func (prodStubAssigneeResolver) ResolveWorkflowApprovalAssignee(context.Context, string, string) (string, error) {
	return "mem_prod", nil
}

// prodAwaitingSeed models a delivered, awaiting send_message.callback effect in
// a running Run whose runnable_at is far in the future so the reducer pass
// never claims it — exactly the frozen-callback parking shape the recovery scan
// targets.
type prodAwaitingSeed struct {
	WorkspaceID       string
	RunID             string
	EffectID          string
	AgentID           string
	Now               time.Time
	FirstNonWorkingAt *time.Time
	NextRecoveryAt    *time.Time
	RecoveryAttempts  int
	// CandidateAgentIDs, when non-empty, makes this a responder.selectOne
	// callback: the durable request carries the ordered candidate pool and an
	// empty responder ref, and the durable receipt/effect froze AgentID as the
	// winner. Recovery must materialize AgentID into the nudge's ResponderAgentRef
	// and clear the pool (the P0 recovery_fix regression).
	CandidateAgentIDs []string
}

func (seed prodAwaitingSeed) mustExec(t *testing.T, ctx context.Context, db coredb.QueryExecer) {
	t.Helper()
	now := seed.Now
	workspaceID, runID, effectID, agentID := seed.WorkspaceID, seed.RunID, seed.EffectID, seed.AgentID
	activationID := "wra_" + effectID
	scopeID := "scp_prod_" + effectID
	messageID := "msg_" + effectID
	input := []byte(`{"seed":"start"}`)
	inputDigest := invocationlang.InputDigest(input)
	var requestJSON string
	if len(seed.CandidateAgentIDs) > 0 {
		// responder.selectOne: durable request carries the ordered candidate pool
		// and an empty responder ref; the durable receipt/effect froze AgentID.
		candidates, candidatesErr := json.Marshal(seed.CandidateAgentIDs)
		require.NoError(t, candidatesErr)
		requestJSON = fmt.Sprintf(`{"intent":{"activationId":"%s","nodeId":"callback","kind":"send_message","callback":true,"responderCandidates":%s,"retry":{"maxAttempts":1,"backoffNanos":1000000000}},"message":{"workspaceId":"%s","planId":"pln_prod","planVersionId":"plv_prod","runId":"%s","activationId":"%s","effectId":"%s","nodeId":"callback","sourceKind":"manual","ownerUserId":"usr_prod","authorMemberId":"mem_prod","clientMessageId":"workflow-run:%s","content":"answer","target":"#prod","callback":true,"responderCandidates":%s}}`,
			activationID, string(candidates), workspaceID, runID, activationID, effectID, effectID, string(candidates))
	} else {
		requestJSON = fmt.Sprintf(`{"intent":{"activationId":"%s","nodeId":"callback","kind":"send_message","callback":true,"responder":"%s","retry":{"maxAttempts":1,"backoffNanos":1000000000}},"message":{"workspaceId":"%s","planId":"pln_prod","planVersionId":"plv_prod","runId":"%s","activationId":"%s","effectId":"%s","nodeId":"callback","sourceKind":"manual","ownerUserId":"usr_prod","authorMemberId":"mem_prod","clientMessageId":"workflow-run:%s","content":"answer","target":"#prod","callback":true,"responderAgentRef":"%s"}}`,
			activationID, agentID, workspaceID, runID, activationID, effectID, effectID, agentID)
	}
	receipt, err := json.Marshal(map[string]any{
		"messageId":                "msg_" + effectID,
		"wakeAgentIds":             []string{agentID},
		"callbackResponderAgentId": agentID,
		"deliveryScope":            map[string]any{"kind": "channel", "scopeId": "chn_prod"},
	})
	require.NoError(t, err)

	mustProdExec(t, ctx, db, `INSERT INTO users (id,email,name,display_name,password_hash,created_at,updated_at) VALUES ('usr_prod','prod@example.com','prod','prod','hash',$1,$1)`, now)
	mustProdExec(t, ctx, db, `INSERT INTO organizations (id,slug,name,created_by_user_id,created_at,updated_at) VALUES ('org_prod','prod','prod','usr_prod',$1,$1)`, now)
	mustProdExec(t, ctx, db, `INSERT INTO workspaces (id,organization_id,slug,name,created_at,updated_at) VALUES ($1,'org_prod','prod','prod',$2,$2)`, workspaceID, now)
	mustProdExec(t, ctx, db, `INSERT INTO members (id,workspace_id,type,display_name,user_id,workspace_role,status,created_at,updated_at) VALUES ('mem_prod',$1,'human','prod','usr_prod','member','active',$2,$2)`, workspaceID, now)
	mustProdExec(t, ctx, db, `INSERT INTO workspace_placements (workspace_id,workspace_group_id,state,placement_epoch,created_at,updated_at) VALUES ($1,'prod-group','active',1,$2,$2)`, workspaceID, now)
	mustProdExec(t, ctx, db, `
INSERT INTO workflow_invocations (
 id,workspace_id,binding_id,binding_revision_id,binding_revision_number,binding_source_id,binding_activation_number,
 plan_id,plan_version_id,workflow_name_snapshot,source_kind,source_occurrence_ref,source_occurred_at,
 input_json,input_digest,source_provenance_json,
 owner_user_id_snapshot,owner_member_id_snapshot,principal_member_id_snapshot,author_member_id_snapshot,
 invoked_by_member_id,invoked_by_actor_type,invoked_by_actor_id,idempotency_key,request_digest,accepted_at
) VALUES (
	$1,$2,NULL,NULL,NULL,NULL,NULL,$3,$4,'Prod Workflow','manual','prod:ref',$5,
	$6,$7,'{}','usr_prod','mem_prod','mem_prod','mem_prod','mem_prod','human','usr_prod','accept-prod',$8,$5
)`, "wiv_"+runID, workspaceID, "pln_prod", "plv_prod", now, input, inputDigest, fmt.Sprintf("%064x", 42))
	mustProdExec(t, ctx, db, `
INSERT INTO workflow_run_instances (
	 id,workspace_id,invocation_id,plan_id,plan_version_id,workflow_name_snapshot,status,
	 input_json,input_digest,state_json,
	 owner_user_id_snapshot,owner_member_id_snapshot,principal_member_id_snapshot,author_member_id_snapshot,
	 queued_at,revision,last_event_seq,runnable_at,created_at,updated_at
) VALUES (
	$1,$2,$3,$4,$5,'Prod Workflow','running',$6,$7,'{}','usr_prod','mem_prod','mem_prod','mem_prod',$8,1,1,$9,$8,$8
)`, runID, workspaceID, "wiv_"+runID, "pln_prod", "plv_prod", input, inputDigest, now, now.Add(24*time.Hour))
	mustProdExec(t, ctx, db, `
INSERT INTO workflow_run_activations (
 id,workspace_id,run_id,logical_key,parent_activation_id,node_id,opcode,status,
 input_json,output_json,scope_context_json,selected_port,attempt_count,retry_at,wake_at,deadline_at,
 error_code,error_json,started_at,completed_at,created_at,updated_at
) VALUES ($1,$2,$3,'cb/prod',NULL,'callback','send_message.callback','waiting',
 '{}',NULL,'{}',NULL,1,NULL,NULL,NULL,NULL,NULL,NULL,NULL,$4,$4)`, activationID, workspaceID, runID, now)
	mustProdExec(t, ctx, db, `
INSERT INTO scopes (id,workspace_id,type,identity_key,status,created_by_member_id,created_at,updated_at)
VALUES ($1,$2,'channel','ch:prod','active','mem_prod',$3,$3)
ON DUPLICATE KEY UPDATE identity_key=identity_key`, scopeID, workspaceID, now)
	mustProdExec(t, ctx, db, `
INSERT INTO scope_messages (id,workspace_id,scope_id,seq,workspace_global_seq,author_member_id,author_type,client_message_id,idempotency_payload_digest,message_type,content,created_at,updated_at)
VALUES ($1,$2,$3,1,1,'mem_prod','human',$4,'0000000000000000000000000000000000000000000000000000000000000099','text','callback',$5,$5)
ON DUPLICATE KEY UPDATE message_type=message_type`, messageID, workspaceID, scopeID, "workflow-run:prod:"+effectID, now)
	mustProdExec(t, ctx, db, `
INSERT INTO workflow_run_effects (
 id,workspace_id,run_id,activation_id,callback_responder_agent_id,message_id,effect_kind,idempotency_key,status,
 request_json,result_json,attempt_count,retry_at,deadline_at,lease_owner,lease_token,lease_expires_at,
 error_code,error_json,delivered_at,completed_at,created_at,updated_at,
 recovery_paused_at,first_non_working_at,recovery_attempt_count,next_recovery_at,
 recovery_cycle_first_non_working_at,recovery_cycle_attempt_count,recovery_cycle_next_recovery_at
) VALUES ($1,$2,$3,$4,$5,$6,'send_message','cb/prod','awaiting',
 $7,$8,0,NULL,NULL,NULL,NULL,NULL,NULL,NULL,$9,NULL,$9,$9,
 NULL,$10,$11,$12,$10,$11,$12)
ON DUPLICATE KEY UPDATE status='awaiting'`,
		effectID, workspaceID, runID, activationID, agentID, messageID, requestJSON, string(receipt), now,
		nullableProdTime(seed.FirstNonWorkingAt), seed.RecoveryAttempts, nullableProdTime(seed.NextRecoveryAt))
	// The awaiting callback still holds its Workflow capacity execution lease
	// (status='held'), exactly the delivered-and-parked production shape. The
	// recovery working judgment does NOT read it (round-3 contract), and the
	// lease reconciler must keep it held while the callback is awaiting.
	mustProdExec(t, ctx, db, `
INSERT INTO workflow_agent_execution_leases (
 workspace_id,responder_agent_id,run_id,activation_id,effect_id,node_id,lease_token,status,created_at,updated_at
) VALUES ($1,$2,$3,$4,$5,'callback','wael_prod','held',$6,$6)
ON DUPLICATE KEY UPDATE effect_id=$5,status='held',updated_at=$6`,
		workspaceID, agentID, runID, activationID, effectID, now)
}

func nullableProdTime(value *time.Time) any {
	if value == nil {
		return nil
	}
	return value.UTC().Truncate(time.Microsecond)
}

func mustProdExec(t *testing.T, ctx context.Context, db coredb.QueryExecer, query string, args ...any) {
	t.Helper()
	_, err := db.Exec(ctx, query, args...)
	require.NoError(t, err)
}

// prodRuntimeWorkingReader is a deterministic orchestrator.RuntimeWorkingReader
// double so the production worker test can drive the canonical runtime-state
// judgment without a live daemon.
type prodRuntimeWorkingReader struct {
	working bool
	known   bool
}

func (reader prodRuntimeWorkingReader) RuntimeWorking(_ context.Context, _ string, _ string) (bool, bool) {
	return reader.working, reader.known
}

// buildProdWorkflowRunWorker composes the exact production worker from wire.go
// (newWorkflowRunWorkerForConfig) with real stores over the given DB and a
// recording message writer. It returns the worker and the writer so the test can
// assert both the recovery outcome and the actual nudge delivered.
func buildProdWorkflowRunWorker(t *testing.T, ctx context.Context, db coredb.DB, runtimeWorking orchestrator.RuntimeWorkingReader) (*orchestrator.Worker, *prodRecordingWriter) {
	t.Helper()
	writer := &prodRecordingWriter{}
	store := run.NewWithOptions(db, run.StoreOptions{})
	worker, err := newWorkflowRunWorkerForConfig(
		config.Config{Server: config.ServerConfig{Profile: config.ServerProfileAll}},
		slog.New(slog.NewTextHandler(io.Discard, nil)),
		store,
		plan.New(db),
		invocation.New(db),
		writer,
		prodStubApprovalWriter{},
		prodStubAssigneeResolver{},
		runtimeWorking,
	)
	require.NoError(t, err)
	require.NotNil(t, worker)
	return worker, writer
}

// TestProductionWorkerTickRecoveryFiresForIdleResponder proves the round-3 P0
// production path: a real server tick (Worker.RunOnce) scans an awaiting
// send_message.callback whose responder is IDLE in the canonical runtime state
// (machine connected, no active turn, no live session), and delivers the frozen
// recovery nudge through the same message writer as the production wire. The
// awaiting callback still holds its Workflow capacity lease, proving the working
// judgment never reads the lease, and the reconciler keeps it held.
func TestProductionWorkerTickRecoveryFiresForIdleResponder(t *testing.T) {
	ctx := context.Background()
	db, cleanup, err := testkit.OpenInitializedTiDBCoreDB(ctx)
	require.NoError(t, err)
	t.Cleanup(func() {
		cleanupCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer cancel()
		require.NoError(t, cleanup(cleanupCtx))
	})
	now := time.Now()
	seed := prodAwaitingSeed{
		WorkspaceID: "wsp_prod_lapse", RunID: "wri_prod_lapse", EffectID: "wfx_prod_lapse", AgentID: "agt_prod_lapse", Now: now,
		FirstNonWorkingAt: timePtr(now.Add(-time.Hour)),
		NextRecoveryAt:    timePtr(now.Add(-time.Minute)),
	}
	seed.mustExec(t, ctx, db)

	// The canonical runtime state says the responder is idle (recoverable).
	worker, writer := buildProdWorkflowRunWorker(t, ctx, db, prodRuntimeWorkingReader{working: false, known: true})
	outcome, err := worker.RunOnce(ctx)
	require.NoError(t, err)

	require.Equal(t, 1, outcome.Recovery.Candidates, "the awaiting callback must be scanned by the production tick")
	require.Equal(t, 1, outcome.Recovery.Sent, "an idle responder must be recovered by the production tick")
	require.Equal(t, 0, outcome.Recovery.Working)
	require.Equal(t, 0, outcome.Recovery.Unknown)
	require.Equal(t, 0, outcome.Recovery.Failed)

	require.Len(t, writer.writes, 1, "the nudge must be written through the production message writer")
	sent := writer.writes[0]
	require.True(t, sent.Callback)
	require.Equal(t, "agt_prod_lapse", sent.ExpectedCallbackResponderAgentID)
	require.Equal(t, "chn_prod", sent.Scope.ScopeID)
	require.Contains(t, sent.Content, "This is automatic recovery attempt 1 of 5")
	require.Contains(t, sent.Content, seed.RunID)
	require.Contains(t, sent.Content, seed.EffectID)

	// The effect advanced to attempt 1 durably.
	var attempts int
	var anchored *time.Time
	err = db.QueryRow(ctx, `SELECT recovery_attempt_count, recovery_cycle_first_non_working_at FROM workflow_run_effects WHERE workspace_id=$1 AND id=$2`, seed.WorkspaceID, seed.EffectID).Scan(&attempts, &anchored)
	require.NoError(t, err)
	require.Equal(t, 1, attempts)
	require.NotNil(t, anchored)

	// Reconciler + queue-lease behavior unchanged: the awaiting callback's held
	// lease is still held (never released by reconcile), even though recovery fired.
	var leaseStatus string
	err = db.QueryRow(ctx, `SELECT status FROM workflow_agent_execution_leases WHERE workspace_id=$1 AND responder_agent_id=$2`, seed.WorkspaceID, seed.AgentID).Scan(&leaseStatus)
	require.NoError(t, err)
	require.Equal(t, "held", leaseStatus)
}

// TestProductionWorkerTickSkipsWorkingResponder proves the corrected canonical
// working-state judgment in the other direction: a responder with a real active
// or queued runtime turn is working, so the production tick scans it but sends
// nothing and starts no timing — even though the callback holds its Workflow
// capacity lease (which must not mask runtime working nor start timing).
// TestProductionWorkerTickRecoverySelectOneMaterializesFrozenWinner proves the
// P0 recovery_fix through the production worker: a responder.selectOne awaiting
// callback whose durable request carries only the candidate pool (empty
// responder ref) and whose receipt/effect froze the winner is recovered by the
// production tick with the frozen winner materialized into the nudge's
// ResponderAgentRef and the candidate pool cleared — no re-selection, no extra
// capacity claim, exactly one responder.
func TestProductionWorkerTickRecoverySelectOneMaterializesFrozenWinner(t *testing.T) {
	ctx := context.Background()
	db, cleanup, err := testkit.OpenInitializedTiDBCoreDB(ctx)
	require.NoError(t, err)
	t.Cleanup(func() {
		cleanupCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer cancel()
		require.NoError(t, cleanup(cleanupCtx))
	})
	now := time.Now()
	seed := prodAwaitingSeed{
		WorkspaceID: "wsp_prod_sel", RunID: "wri_prod_sel", EffectID: "wfx_prod_sel", AgentID: "agt_sel",
		Now: now, FirstNonWorkingAt: timePtr(now.Add(-time.Hour)), NextRecoveryAt: timePtr(now.Add(-time.Minute)),
		CandidateAgentIDs: []string{"agt_sel", "agt_other"},
	}
	seed.mustExec(t, ctx, db)

	worker, writer := buildProdWorkflowRunWorker(t, ctx, db, prodRuntimeWorkingReader{working: false, known: true})
	outcome, err := worker.RunOnce(ctx)
	require.NoError(t, err)

	require.Equal(t, 1, outcome.Recovery.Candidates)
	require.Equal(t, 1, outcome.Recovery.Sent, "the selectOne callback must be recovered by the production tick")
	require.Equal(t, 0, outcome.Recovery.Failed)
	require.Len(t, writer.writes, 1)

	sent := writer.writes[0]
	require.Equal(t, "agt_sel", sent.ResponderAgentRef,
		"the recovery nudge must materialize the durable frozen winner")
	require.Empty(t, sent.ResponderCandidates,
		"the recovery nudge must clear the candidate pool (no re-selection)")
	require.Equal(t, "agt_sel", sent.ExpectedCallbackResponderAgentID)
	require.True(t, sent.Callback)
}

func TestProductionWorkerTickSkipsWorkingResponder(t *testing.T) {
	ctx := context.Background()
	db, cleanup, err := testkit.OpenInitializedTiDBCoreDB(ctx)
	require.NoError(t, err)
	t.Cleanup(func() {
		cleanupCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer cancel()
		require.NoError(t, cleanup(cleanupCtx))
	})
	now := time.Now()
	seed := prodAwaitingSeed{
		WorkspaceID: "wsp_prod_work", RunID: "wri_prod_work", EffectID: "wfx_prod_work", AgentID: "agt_prod_work", Now: now,
	}
	seed.mustExec(t, ctx, db)

	worker, writer := buildProdWorkflowRunWorker(t, ctx, db, prodRuntimeWorkingReader{working: true, known: true})
	outcome, err := worker.RunOnce(ctx)
	require.NoError(t, err)

	require.Equal(t, 1, outcome.Recovery.Candidates)
	require.Equal(t, 1, outcome.Recovery.Working, "a working responder must be skipped")
	require.Equal(t, 0, outcome.Recovery.Sent)
	require.Len(t, writer.writes, 0, "working callbacks must never be nudged")

	// No timing started durably.
	var firstNonWorking any
	err = db.QueryRow(ctx, `SELECT recovery_cycle_first_non_working_at FROM workflow_run_effects WHERE workspace_id=$1 AND id=$2`, seed.WorkspaceID, seed.EffectID).Scan(&firstNonWorking)
	require.NoError(t, err)
	require.Nil(t, firstNonWorking)

	// Queue-lease behavior unchanged: the live owner's lease is still held.
	var leaseStatus string
	err = db.QueryRow(ctx, `SELECT status FROM workflow_agent_execution_leases WHERE workspace_id=$1 AND responder_agent_id=$2`, seed.WorkspaceID, seed.AgentID).Scan(&leaseStatus)
	require.NoError(t, err)
	require.Equal(t, "held", leaseStatus)
}

// TestProductionWorkerTickUnknownDoesNotSend proves the Leader round-3 fail-safe:
// when the canonical runtime state is unknown (e.g. machine not reconnected after
// restart), the production tick scans but never sends and never starts timing.
func TestProductionWorkerTickUnknownDoesNotSend(t *testing.T) {
	ctx := context.Background()
	db, cleanup, err := testkit.OpenInitializedTiDBCoreDB(ctx)
	require.NoError(t, err)
	t.Cleanup(func() {
		cleanupCtx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
		defer cancel()
		require.NoError(t, cleanup(cleanupCtx))
	})
	now := time.Now()
	seed := prodAwaitingSeed{
		WorkspaceID: "wsp_prod_unk", RunID: "wri_prod_unk", EffectID: "wfx_prod_unk", AgentID: "agt_prod_unk", Now: now,
	}
	seed.mustExec(t, ctx, db)

	worker, writer := buildProdWorkflowRunWorker(t, ctx, db, prodRuntimeWorkingReader{working: false, known: false})
	outcome, err := worker.RunOnce(ctx)
	require.NoError(t, err)

	require.Equal(t, 1, outcome.Recovery.Candidates)
	require.Equal(t, 1, outcome.Recovery.Unknown, "an unknown responder must be skipped without sending")
	require.Equal(t, 0, outcome.Recovery.Sent)
	require.Len(t, writer.writes, 0, "unknown state must never trigger a nudge")
}

func timePtr(value time.Time) *time.Time { return &value }