Agent-System-Backend/internal/server/run_process_events_test.go

package server

import (
	"encoding/json"
	"net/http"
	"strconv"
	"testing"
	"time"

	"agent-system-backend/internal/store"
	core "agent-system-core"
)

func TestRunProcessEventsAreOrderedAndBoundToConversation(t *testing.T) {
	srv, st := newTestServer(t)
	h := srv.Handler()
	_, convID := setupConversation(t, h)
	_, otherConvID := setupConversation(t, h)
	runID := postMessage(t, h, convID, "整理材料")
	waitRunTerminal(t, h, runID, 5*time.Second)
	events := []core.StreamEvent{
		{RunID: runID, Type: core.EvThinkingDelta, Payload: json.RawMessage(`{"index":0,"text":"先核对日期"}`)},
		{RunID: runID, Type: core.EvToolOutput, Payload: json.RawMessage(`{"callId":"call-1","stream":"stdout","text":"完整工具输出"}`)},
	}
	if _, err := st.AppendEvents(convID, events); err != nil {
		t.Fatal(err)
	}
	// The runner may persist its terminal event after its status becomes terminal.
	// Anchor paging to the events this test appended, not an earlier snapshot.
	path := "/api/conversations/" + convID + "/runs/" + runID + "/events?after=" + strconv.FormatInt(events[0].Seq-1, 10) + "&limit=1"
	code, first := doJSON(t, h, http.MethodGet, path, nil)
	if code != http.StatusOK {
		t.Fatalf("first page: %d %v", code, first)
	}
	page := first["events"].([]any)
	if len(page) != 1 || page[0].(map[string]any)["type"] != string(core.EvThinkingDelta) {
		t.Fatalf("first page: %v", page)
	}
	seq := int64(page[0].(map[string]any)["seq"].(float64))
	code, second := doJSON(t, h, http.MethodGet, "/api/conversations/"+convID+"/runs/"+runID+"/events?after="+strconv.FormatInt(seq, 10)+"&limit=1", nil)
	if code != http.StatusOK {
		t.Fatalf("second page: %d %v", code, second)
	}
	page = second["events"].([]any)
	if len(page) != 1 || page[0].(map[string]any)["type"] != string(core.EvToolOutput) {
		t.Fatalf("second page: %v", page)
	}
	if payload := page[0].(map[string]any)["payload"].(map[string]any); payload["text"] != "完整工具输出" {
		t.Fatalf("payload: %v", payload)
	}
	parent, err := st.GetRunByID(runID)
	if err != nil {
		t.Fatal(err)
	}
	child := core.Run{ID: "run_child_process", ParentRunID: parent.ID, ConversationID: parent.ConversationID,
		ProjectID: parent.ProjectID, Owner: parent.Owner, Kind: core.RunSubagent, Status: core.RunCompleted}
	if err := st.CreateRun(&child); err != nil {
		t.Fatal(err)
	}
	childEvents := []core.StreamEvent{
		{RunID: child.ID, Type: core.EvAgentProgress, Payload: json.RawMessage(`{"agentRun":"run_child_process","note":"正在核对附录"}`)},
		{RunID: child.ID, Type: core.EvToolOutput, Payload: json.RawMessage(`{"callId":"hidden","text":"child raw output"}`)},
		{RunID: child.ID, Type: core.EvAgentResult, Payload: json.RawMessage(`{"agentRun":"run_child_process","status":"completed","summary":"附录已核对"}`)},
	}
	if _, err := st.AppendEvents(convID, childEvents); err != nil {
		t.Fatal(err)
	}
	code, childPage := doJSON(t, h, http.MethodGet, "/api/conversations/"+convID+"/runs/"+runID+"/events?after="+strconv.FormatInt(events[1].Seq, 10), nil)
	if code != http.StatusOK {
		t.Fatalf("child page: %d %v", code, childPage)
	}
	childRows := childPage["events"].([]any)
	if len(childRows) != 2 || childRows[0].(map[string]any)["type"] != string(core.EvAgentProgress) || childRows[1].(map[string]any)["type"] != string(core.EvAgentResult) {
		t.Fatalf("child progress/result not retained or raw child output leaked: %v", childRows)
	}
	code, _ = doJSON(t, h, http.MethodGet, "/api/conversations/"+otherConvID+"/runs/"+runID+"/events", nil)
	if code != http.StatusNotFound {
		t.Fatalf("mismatched conversation exposed run: %d", code)
	}
}

func TestRunProcessEventsRejectsDisabledImplicitLocal(t *testing.T) {
	first, second := twinRecoveryStores(t)
	t.Logf("process-event replay dialect=%s independent stores", first.Dialect())
	if err := first.CreateAccount("admin", "", "", true); err != nil {
		t.Fatal(err)
	}
	if err := first.EnsureUser(store.OwnerDefault, ""); err != nil {
		t.Fatal(err)
	}
	project, err := first.CreateProject(store.OwnerDefault, "Project")
	if err != nil {
		t.Fatal(err)
	}
	conversation, err := first.CreateConversation(store.OwnerDefault, project.ID, "Process", "")
	if err != nil {
		t.Fatal(err)
	}
	run := core.Run{ID: "run_local_process", ConversationID: conversation.ID, ProjectID: project.ID,
		Owner: store.OwnerDefault, Kind: core.RunConversation, Status: core.RunCompleted}
	if err := first.CreateRun(&run); err != nil {
		t.Fatal(err)
	}
	if _, err := first.AppendEvents(conversation.ID, []core.StreamEvent{{RunID: run.ID, Type: core.EvThinkingDelta,
		Payload: json.RawMessage(`{"index":0,"text":"private process"}`)}}); err != nil {
		t.Fatal(err)
	}
	srv := newAuthTestServerOnStore(t, second, t.TempDir())
	h := srv.Handler()
	path := "/api/conversations/" + conversation.ID + "/runs/" + run.ID + "/events"
	code, _ := doJSON(t, h, http.MethodGet, path, nil)
	if code != http.StatusUnauthorized {
		t.Fatalf("auth-on replay without credential: %d", code)
	}
	srv.Auth.Enabled = false
	code, _ = doJSON(t, h, http.MethodGet, path, nil)
	if code != http.StatusOK {
		t.Fatalf("enabled local account could not read process events: %d", code)
	}
	account, err := first.GetAccountByOwner(store.OwnerDefault)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := first.SetAccountEnabled("admin", account.ID, account.StateVersion, false); err != nil {
		t.Fatal(err)
	}
	code, _ = doJSON(t, h, http.MethodGet, path, nil)
	if code != http.StatusForbidden {
		t.Fatalf("disabled local account read process events: %d", code)
	}
}