Agent-System-Backend/internal/runtime/workflow_compiled_test.go

package runtime

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"

	agentllm "agent-native/runtime/llm"
	"agent-system-backend/internal/projectcontext"
	"agent-system-backend/internal/workflow"
	"agent-system-backend/pkg/extension"
	core "agent-system-core"
)

// workflow_compiled_test.go — Q3 WP2 行为一致性（P4）：run_workflow 三分支
// 改投编译图后的 runtime 级验证——注册工作流查找与错误面、规划分支确认门
// 与动态上限、模型脚本开关、缺失槽位 HITL 收齐、验收模型路由。
// （编译/宿主语义的细粒度断言在 internal/wfcompile 测试内。）

type wfFake struct {
	ctx       context.Context
	launch    workflow.Launch
	hitl      []core.HITLResponse
	hitlQ     []string
	plans     []core.PlanSnapshot
	tools     []core.ToolCallRequest
	prompts   []string
	writes    []string
	cards     []core.ActivityCard
	assistant []core.Part
	arts      map[string]core.Artifact
	byID      map[string]core.WorkflowDef
	spawnText string
	spawnIDs  []string
	toolNames []string
	settings  core.EffectiveSettings
	sandbox   string
	scripts   map[string]string
}

func (f *wfFake) Context() context.Context {
	if f.ctx == nil {
		return context.Background()
	}
	return f.ctx
}
func (f *wfFake) Run() core.Run              { return core.Run{ID: "run_wf"} }
func (f *wfFake) Profile() core.AgentProfile { return core.AgentProfile{} }
func (f *wfFake) Settings() core.EffectiveSettings {
	if f.settings == nil {
		return core.EffectiveSettings{}
	}
	return f.settings
}
func (f *wfFake) ConversationSandbox() string { return f.sandbox }
func (f *wfFake) Plan() (core.PlanSnapshot, bool) {
	if len(f.plans) == 0 {
		return core.PlanSnapshot{}, false
	}
	p := f.plans[len(f.plans)-1]
	p.Steps = append([]core.PlanStep(nil), p.Steps...)
	return p, true
}
func (f *wfFake) WorkflowScript(id, path string) (string, bool) {
	body, ok := f.scripts[workflow.ScriptKey(id, path)]
	return body, ok
}
func (f *wfFake) Emit(core.StreamEventType, any) error { return nil }
func (f *wfFake) SetAssistantParts(parts []core.Part) error {
	f.assistant = parts
	return nil
}
func (f *wfFake) AskHITL(card core.HITLCard) (core.HITLResponse, error) {
	f.hitlQ = append(f.hitlQ, card.Question)
	if len(f.hitl) == 0 {
		return core.HITLResponse{}, context.Canceled
	}
	resp := f.hitl[0]
	f.hitl = f.hitl[1:]
	return resp, nil
}
func (f *wfFake) SpawnSubagent(name, task, hint string) (core.AgentResultPayload, error) {
	return f.SpawnAgent("agent.sub.general", name, task, hint)
}
func (f *wfFake) SpawnAgent(agentID, name, task, hint string) (core.AgentResultPayload, error) {
	f.prompts = append(f.prompts, agentID+"|"+task)
	f.spawnIDs = append(f.spawnIDs, agentID)
	text := f.spawnText
	if text == "" {
		text = "正文 art_0123456789ab"
	}
	return core.AgentResultPayload{Text: text, Status: core.RunCompleted}, nil
}
func (f *wfFake) UpdatePlan(p core.PlanSnapshot) error {
	p.Steps = append([]core.PlanStep(nil), p.Steps...)
	f.plans = append(f.plans, p)
	return nil
}
func (f *wfFake) UpdateTodos(core.TodoList) error { return nil }
func (f *wfFake) WriteArtifact(name, kind, mime string, content []byte) (core.Artifact, error) {
	f.writes = append(f.writes, name+"="+string(content))
	art := core.Artifact{ID: "art_aaaaaaaaaaaa", Name: name, Kind: kind}
	if f.arts == nil {
		f.arts = map[string]core.Artifact{}
	}
	f.arts[string(art.ID)] = art
	return art, nil
}
func (f *wfFake) Retrieve(core.RetrievalQuery) ([]core.RetrievalHit, error) { return nil, nil }
func (f *wfFake) ExecuteTool(call core.ToolCallRequest) core.ToolCallResult {
	f.tools = append(f.tools, call)
	if call.Name == "make_file" {
		art := core.Artifact{ID: "art_bbbbbbbbbbbb", Name: "made.md"}
		if f.arts == nil {
			f.arts = map[string]core.Artifact{}
		}
		f.arts[string(art.ID)] = art
		return core.ToolCallResult{Status: "ok", Output: "wrote", Artifact: &art}
	}
	return core.ToolCallResult{Status: "ok", Output: "12:00"}
}
func (f *wfFake) History() []core.Message            { return nil }
func (f *wfFake) Attachments() []core.ParsedDocument { return nil }
func (f *wfFake) Tools() []core.ToolDesc {
	var out []core.ToolDesc
	for _, name := range f.toolNames {
		out = append(out, core.ToolDesc{Name: name})
	}
	return out
}
func (f *wfFake) Subtask() string { return "" }
func (f *wfFake) CreateCard(card core.ActivityCard) (core.ActivityCard, error) {
	card.ID = "card_1"
	f.cards = append(f.cards, card)
	return card, nil
}
func (f *wfFake) SetCardStatus(string, core.CardStatus, string, string) error { return nil }
func (f *wfFake) SubmitCard(card core.ActivityCard) (core.ActivityCard, error) {
	return card, nil
}
func (f *wfFake) BeginStreamCard(card core.ActivityCard) (core.ActivityCard, error) {
	return card, nil
}
func (f *wfFake) AppendStreamCard(string) error                { return nil }
func (f *wfFake) CommitStreamCard() (core.ActivityCard, error) { return core.ActivityCard{}, nil }
func (f *wfFake) Card(string) *core.ActivityCard               { return nil }
func (f *wfFake) Artifact(id string) (core.Artifact, []byte, bool) {
	if f.arts == nil {
		f.arts = map[string]core.Artifact{}
	}
	if id == "art_0123456789ab" {
		return core.Artifact{ID: core.ID(id), Name: "note.md"}, []byte("x"), true
	}
	art, ok := f.arts[id]
	return art, nil, ok
}
func (f *wfFake) SetContextObservation(core.ContextObservation) {}
func (f *wfFake) Workflow() (workflow.Launch, bool)             { return f.launch, true }
func (f *wfFake) WorkflowByID(id string) (core.WorkflowDef, bool) {
	def, ok := f.byID[id]
	return def, ok
}

// ---- run_workflow 注册分支：查找与错误面（legacy 同款文案） ----

func TestRunWorkflowToolUsesRegisteredDefinition(t *testing.T) {
	f := &wfFake{hitl: []core.HITLResponse{{ChoiceID: "yes"}}}
	f.byID = map[string]core.WorkflowDef{
		"demo.brief": {ID: "demo.brief", Title: "简报", Steps: []core.WorkflowStep{
			{ID: "ask", Title: "确认", Kind: "hitl", Hitl: &core.HITLCardTemplate{Question: "可以吗？", Kind: core.HITLKindConfirm}},
		}},
	}
	st := &directRun{rc: f}
	out := st.runRegisteredWorkflow(`{"workflow":"demo.brief","inputs":{"topic":"核能"}}`)
	if !strings.Contains(out, "完成") || f.hitlQ[0] != "可以吗？" {
		t.Fatalf("out %s questions %v", out, f.hitlQ)
	}
	if got := st.runRegisteredWorkflow(`{"workflow":"missing"}`); !strings.Contains(got, "未注册") {
		t.Fatalf("%s", got)
	}
	if got := st.runRegisteredWorkflow(`{"goal":"做一件事","workflow":"demo.brief"}`); !strings.Contains(got, "只能") {
		t.Fatalf("%s", got)
	}
}

// ---- 规划分支：确认门（engine confirmation）+ 动态上限 fail-closed ----

func TestPlannedWorkflowNeedsConfirm(t *testing.T) {
	plan := `{"id":"plan.brief","title":"简报","steps":[{"id":"ask","title":"确认","kind":"hitl","hitl":{"question":"可以吗？","kind":"confirm","allowFreeInput":false}}]}`
	f := &wfFake{
		spawnText: plan,
		hitl:      []core.HITLResponse{{ChoiceID: "approve"}, {ChoiceID: "ok"}},
	}
	st := &directRun{rc: f}
	out := st.runRegisteredWorkflow(`{"goal":"写一份简报"}`)
	if f.spawnIDs[0] != "agent.sub.workflow-plan" || !strings.Contains(f.hitlQ[0], "确认") {
		t.Fatalf("spawn %v questions %v", f.spawnIDs, f.hitlQ)
	}
	if !strings.Contains(out, "完成") || f.hitlQ[1] != "可以吗？" {
		t.Fatalf("out %s questions %v", out, f.hitlQ)
	}

	// 计划确认被拒 → legacy 语义「未确认，未执行」。
	f2 := &wfFake{spawnText: plan, hitl: []core.HITLResponse{{ChoiceID: "abort"}}}
	out = (&directRun{rc: f2}).runRegisteredWorkflow(`{"goal":"写一份简报"}`)
	if out != "未确认，未执行" || len(f2.hitlQ) != 1 {
		t.Fatalf("out %s questions %v", out, f2.hitlQ)
	}

	// 动态上限：9 步 > 8，编译入口拒绝，不进任何 HITL。
	var steps []string
	for i := 0; i < 9; i++ {
		steps = append(steps, fmt.Sprintf(`{"id":"s%d","title":"步","kind":"hitl"}`, i))
	}
	f3 := &wfFake{spawnText: `{"id":"too.big","title":"太多","steps":[` + strings.Join(steps, ",") + `]}`}
	out = (&directRun{rc: f3}).runRegisteredWorkflow(`{"goal":"做很多事"}`)
	if !strings.Contains(out, "超出上限") || len(f3.hitlQ) != 0 {
		t.Fatalf("out %s questions %v", out, f3.hitlQ)
	}
}

// ---- 脚本分支：runtime.workflowScripts 开关（默认关） ----

func TestScriptBranchRespectsSwitch(t *testing.T) {
	st := &directRun{rc: &wfFake{sandbox: t.TempDir()}}
	out := st.runScriptWorkflow("import agentwf as wf\nwf.step('run','跑','done')\n")
	if !strings.Contains(out, "工作流脚本未开启") {
		t.Fatalf("开关默认关应失败：%s", out)
	}
}

// ---- 缺失槽位：执行前 HITL 收齐并参与编译期渲染 ----

func TestCompiledRunnerAsksMissingSlot(t *testing.T) {
	f := &wfFake{hitl: []core.HITLResponse{{FreeText: "核能"}, {ChoiceID: "short"}}}
	f.launch = workflow.Launch{
		Missing: []workflow.SlotQuestion{{Key: "topic", Question: "主题是什么？"}},
		Def: core.WorkflowDef{ID: "slots", Title: "槽位", Steps: []core.WorkflowStep{
			{ID: "len", Title: "篇幅", Kind: "hitl", Hitl: &core.HITLCardTemplate{Question: "篇幅？", Kind: core.HITLKindChoice, Choices: []core.HITLChoice{{ID: "short", Label: "短"}}}},
			{ID: "save", Title: "存稿", Kind: "artifact", Artifact: "brief.md", Prompt: "主题：{{slots.topic}}，篇幅：{{steps.len.output}}"},
		}},
	}
	if err := (CompiledWorkflowRunner{}).Execute(f); err != nil {
		t.Fatal(err)
	}
	if f.hitlQ[0] != "主题是什么？" || f.hitlQ[1] != "篇幅？" {
		t.Fatalf("questions %v", f.hitlQ)
	}
	if len(f.writes) != 1 || !strings.HasPrefix(f.writes[0], "brief.md=") ||
		!strings.Contains(f.writes[0], "主题：核能") || !strings.Contains(f.writes[0], "short") {
		t.Fatalf("writes = %v", f.writes)
	}
	text := f.assistant[0].Text
	if !strings.Contains(text, "工作流「槽位」完成。") {
		t.Fatalf("summary %s", text)
	}
}

// ---- 验收模型路由：轻量模型配置 → JudgeStream；未配置 → nil（fail-closed） ----

func TestJudgeStreamUsesConfiguredModel(t *testing.T) {
	var got string
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var body map[string]any
		_ = json.NewDecoder(r.Body).Decode(&body)
		got, _ = body["model"].(string)
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"{\\\"passed\\\":true,\\\"reason\\\":\\\"ok\\\",\\\"nextAction\\\":\\\"\\\"}\"}}]}\n\n")
		_, _ = io.WriteString(w, "data: [DONE]\n\n")
	}))
	t.Cleanup(srv.Close)
	f := &wfFake{settings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{
		"baseUrl": srv.URL, "apiKey": "test-key", "model": "MiniMax-test",
	}}}}
	stream := judgeStreamFor(f)
	if stream == nil {
		t.Fatal("已配置轻量模型时必须给出验收流")
	}
	for ev := range stream(context.Background(), "verify", []agentllm.Message{{Role: agentllm.RoleUser, Content: "x"}}, nil, agentllm.StreamOptions{}) {
		if ev.Kind == agentllm.KindError {
			t.Fatal(ev.Text)
		}
	}
	if got != "MiniMax-test" {
		t.Fatalf("model %q", got)
	}
	if judgeStreamFor(&wfFake{}) != nil {
		t.Fatal("未配置模型应为 nil（验证器不可用 fail-closed）")
	}
}

type projectJudgeCtx struct {
	*wfFake
	manifests []extension.ProjectContextManifest
	authEpoch string
}

func (c *projectJudgeCtx) Run() core.Run {
	return core.Run{ID: "run-judge", Owner: "alice", ProjectID: "project-1"}
}
func (c *projectJudgeCtx) ProjectContext(context.Context, string, int) (extension.ProjectContextBundle, bool, error) {
	epoch := c.authEpoch
	if epoch == "" {
		epoch = "duty-1"
	}
	return extension.ProjectContextBundle{Status: extension.ProjectContextReady, ProjectID: "project-1", AuthorizationEpoch: epoch, PublicationEpoch: "pub-1", Items: []extension.ProjectContextItem{{SourceID: "judge-source", Revision: "r1", Hash: "h1", Text: "current project rule", Required: true}}}, true, nil
}
func (c *projectJudgeCtx) SaveContextManifest(_ context.Context, m extension.ProjectContextManifest) error {
	c.manifests = append(c.manifests, m)
	return nil
}
func (c *projectJudgeCtx) ValidateProjectCrossGuard(_ context.Context, current extension.ProjectContextBundle) error {
	for _, m := range c.manifests {
		if m.AuthorizationEpoch != current.AuthorizationEpoch || m.PublicationEpoch != current.PublicationEpoch || m.Watermark != current.Watermark {
			return fmt.Errorf("prior workflow model source was revoked")
		}
	}
	return nil
}

func TestWorkflowJudgeRequestUsesProjectContext(t *testing.T) {
	var sent []string
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var body struct {
			Messages []struct {
				Content string `json:"content"`
			} `json:"messages"`
		}
		_ = json.NewDecoder(r.Body).Decode(&body)
		var input string
		for _, m := range body.Messages {
			input += m.Content
		}
		sent = append(sent, input)
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = io.WriteString(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n")
		_, _ = io.WriteString(w, "data: [DONE]\n\n")
	}))
	defer srv.Close()
	c := &projectJudgeCtx{wfFake: &wfFake{settings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"baseUrl": srv.URL, "apiKey": "test-key", "model": "test-model"}}}}}
	stream := judgeStreamFor(c)
	for ev := range stream(context.Background(), "test-model", []agentllm.Message{{Role: agentllm.RoleUser, Content: "judge"}}, nil, agentllm.StreamOptions{}) {
		if ev.Kind == agentllm.KindError {
			t.Fatalf("judge stream failed: %+v", ev)
		}
	}
	second := []agentllm.Message{{Role: agentllm.RoleSystem, Content: "judge role"}, {Role: agentllm.RoleUser, Content: "judge"}, {Role: agentllm.RoleAssistant, Content: "previous authorized step"}, {Role: agentllm.RoleTool, Content: "NEW_AUTHORIZED_TOOL"}}
	for ev := range stream(context.Background(), "test-model", second, nil, agentllm.StreamOptions{}) {
		if ev.Kind == agentllm.KindError {
			t.Fatalf("second judge stream failed: %+v", ev)
		}
	}
	if len(sent) != 2 || !strings.Contains(sent[0], "current project rule") || !strings.Contains(sent[1], "NEW_AUTHORIZED_TOOL") || !strings.Contains(sent[1], "current project rule") || len(c.manifests) != 4 || c.manifests[3].Delivery != "responded" {
		t.Fatalf("workflow judge missed project context: input=%q manifests=%+v", sent, c.manifests)
	}
}

func TestWorkflowModelReceiptCarriesEngineStepAttempt(t *testing.T) {
	c := &projectJudgeCtx{wfFake: &wfFake{}, authEpoch: "duty-1"}
	ctx := projectcontext.WithWorkflowStepAttempt(context.Background(), "engine-run/draft/2")
	for ev := range GuardProjectStream(c, "draft", scriptStream(textTurns("done"), nil))(ctx, "model", []agentllm.Message{{Role: agentllm.RoleUser, Content: "write"}}, nil, agentllm.StreamOptions{}) {
		if ev.Kind == agentllm.KindError {
			t.Fatalf("step model failed: %+v", ev)
		}
	}
	if len(c.manifests) != 2 || c.manifests[0].WorkflowStepAttemptID != "engine-run/draft/2" || c.manifests[1].WorkflowStepAttemptID != "engine-run/draft/2" || c.manifests[1].Delivery != "responded" {
		t.Fatalf("engine attempt missing from trusted model receipts: %+v", c.manifests)
	}
}

func TestWorkflowJudgeBlocksOutputFromEarlierAuthorizationEpoch(t *testing.T) {
	const old = "OLD_RESTRICTED_BODY"
	seen := false
	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		var body struct {
			Messages []struct {
				Content string `json:"content"`
			} `json:"messages"`
		}
		_ = json.NewDecoder(r.Body).Decode(&body)
		for _, m := range body.Messages {
			if strings.Contains(m.Content, old) {
				seen = true
			}
		}
		w.Header().Set("Content-Type", "text/event-stream")
		_, _ = io.WriteString(w, "data: [DONE]\n\n")
	}))
	defer srv.Close()
	c := &projectJudgeCtx{wfFake: &wfFake{settings: core.EffectiveSettings{"models": map[string]any{"chat": map[string]any{"baseUrl": srv.URL, "apiKey": "test-key", "model": "test-model"}}}}, authEpoch: "duty-1"}
	step := GuardProjectStream(c, "write", scriptStream(textTurns("step done"), nil))
	for range step(context.Background(), "test-model", []agentllm.Message{{Role: agentllm.RoleUser, Content: "write report"}}, nil, agentllm.StreamOptions{}) {
	}
	if len(c.manifests) != 2 || c.manifests[1].Delivery != "responded" {
		t.Fatalf("step source receipt missing: %+v", c.manifests)
	}
	c.authEpoch = "duty-2" // member remains; document duty and publication changed before critic
	blocked := false
	for ev := range judgeStreamFor(c)(context.Background(), "test-model", []agentllm.Message{{Role: agentllm.RoleUser, Content: "Judge output.md:\n" + old}}, nil, agentllm.StreamOptions{}) {
		blocked = blocked || ev.Kind == agentllm.KindError
	}
	if seen || !blocked {
		t.Fatalf("old workflow output reached a separate judge guard: seen=%v blocked=%v", seen, blocked)
	}
}

func (*wfFake) Surfaces() []core.SurfaceSnapshot { return nil }