Agent-System-Backend/internal/app/project_work_observation_test.go

package app

import (
	"context"
	"crypto/sha256"
	"fmt"
	"path/filepath"
	"testing"

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

func workObserverForTest() extension.ProjectWorkObserver {
	return extension.ProjectWorkObserver{
		Observe: func(context.Context, extension.WorkObservation) (extension.WorkObservationResult, error) {
			return extension.WorkObservationResult{State: extension.WorkObservationPending}, nil
		},
		Preflight: func(context.Context, extension.ProjectWorkAction) (extension.ProjectWorkDecision, error) {
			return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
		},
		Respond: func(context.Context, extension.ProjectRiskAnswer) (extension.ProjectWorkDecision, error) {
			return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
		},
	}
}

func TestProjectWorkDeliveryRechecksCurrentAuthorityAndACKsOnce(t *testing.T) {
	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "work.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()
	if err := st.EnsureUser("alice", "Alice"); err != nil {
		t.Fatal(err)
	}
	project, err := st.CreateProject("alice", "P")
	if err != nil {
		t.Fatal(err)
	}
	conv, err := st.CreateConversation("alice", string(project.ID), "Work", "")
	if err != nil {
		t.Fatal(err)
	}
	msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "要求效率和成本都低于人工"}}}
	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &msg, "req-1"); err != nil {
		t.Fatal(err)
	}
	called := 0
	slot := &projectWorkObserverSlot{}
	observer := workObserverForTest()
	observer.Observe = func(_ context.Context, source extension.WorkObservation) (extension.WorkObservationResult, error) {
		called++
		if source.Owner != "alice" || source.Actor != "alice" || source.ProjectID != project.ID || source.SourceID != string(msg.ID) || source.Text != msg.Parts[0].Text || source.Scope != "private" {
			t.Fatalf("provider received forged or wrong source: %+v", source)
		}
		return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: "rdm-receipt-1"}, nil
	}
	if err := (pluginRegistry{pluginID: "rdm", caps: hostcaps{workObserver: slot}}).RegisterProjectWorkObserver(observer); err != nil {
		t.Fatal(err)
	}
	denied := func(context.Context, string, core.ID) (bool, error) { return false, nil }
	if _, err := deliverProjectWorkObservations(context.Background(), st, slot, denied, 10); err == nil || called != 0 {
		t.Fatalf("revoked source was disclosed: calls=%d err=%v", called, err)
	}
	allowed := func(_ context.Context, actor string, id core.ID) (bool, error) {
		return actor == "alice" && id == project.ID, nil
	}
	if delivered, err := deliverProjectWorkObservations(context.Background(), st, slot, allowed, 10); err != nil || delivered != 1 || called != 1 {
		t.Fatalf("authorized source delivery failed: delivered=%d calls=%d err=%v", delivered, called, err)
	}
	if delivered, err := deliverProjectWorkObservations(context.Background(), st, slot, allowed, 10); err != nil || delivered != 0 || called != 1 {
		t.Fatalf("delivered source was replayed without need: delivered=%d calls=%d err=%v", delivered, called, err)
	}
}

func TestProjectMessagePreflightRequiresAnalysisAndExactManifest(t *testing.T) {
	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "preflight.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()
	if err := st.EnsureUser("alice", "Alice"); err != nil {
		t.Fatal(err)
	}
	project, err := st.CreateProject("alice", "P")
	if err != nil {
		t.Fatal(err)
	}
	conv, err := st.CreateConversation("alice", string(project.ID), "Work", "")
	if err != nil {
		t.Fatal(err)
	}
	msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "要求效率和成本都低于人工"}}}
	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &msg, "req-1"); err != nil {
		t.Fatal(err)
	}
	ready, observations, decisions := false, 0, 0
	observer := workObserverForTest()
	observer.Observe = func(context.Context, extension.WorkObservation) (extension.WorkObservationResult, error) {
		observations++
		return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: "intake-1"}, nil
	}
	observer.Preflight = func(_ context.Context, action extension.ProjectWorkAction) (extension.ProjectWorkDecision, error) {
		decisions++
		if action.Actor != "alice" || action.ProjectID != project.ID || action.ConversationID != conv.ID || action.RunID != "" || action.Action != "conversation.run" || len(action.Manifest) != 1 {
			t.Fatalf("preflight did not receive exact trusted source manifest: %+v", action)
		}
		if !ready {
			return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
		}
		return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "decision-1", SourceWatermark: action.SourceWatermark}, nil
	}
	slot := &projectWorkObserverSlot{provider: &observer}
	work := func(_ context.Context, actor string, id core.ID) (bool, error) {
		return actor == "alice" && id == project.ID, nil
	}
	decision, err := preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
	if err != nil || decision.State != extension.ProjectWorkPending || observations != 1 || decisions != 1 {
		t.Fatalf("durable intake was confused with completed analysis: %+v observations=%d decisions=%d err=%v", decision, observations, decisions, err)
	}
	ready = true
	decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
	if err != nil || decision.State != extension.ProjectWorkProceed || observations != 1 || decisions != 2 {
		t.Fatalf("same source did not proceed after domain analysis: %+v observations=%d decisions=%d err=%v", decision, observations, decisions, err)
	}
	if _, err := st.DB().Exec(`UPDATE messages SET parts=? WHERE id=?`, `[{"type":"text","text":"changed"}]`, msg.ID); err != nil {
		t.Fatal(err)
	}
	decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(msg.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
	if err != nil || decision.State != extension.ProjectWorkPending || decisions != 2 {
		t.Fatalf("source changed after ACK but stale decision still passed: %+v decisions=%d err=%v", decision, decisions, err)
	}
	partial := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartArtifactRef, Artifact: &core.ArtifactRef{ArtifactID: "file-1"}}}}
	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &partial, "req-2"); err != nil {
		t.Fatal(err)
	}
	decision, err = preflightProjectMessage(context.Background(), st, slot, work, string(partial.ID), "conversation.run", "alice", string(conv.ID), string(project.ID), "conversation.run", "v1")
	if err != nil || decision.State != extension.ProjectWorkPending || decisions != 2 {
		t.Fatalf("partial attachment reached project action preflight: %+v decisions=%d err=%v", decision, decisions, err)
	}
}

func TestRetrievalObservationRechecksCurrentSourceBeforeDisclosure(t *testing.T) {
	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "source.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()
	if err := st.EnsureUser("alice", "Alice"); err != nil {
		t.Fatal(err)
	}
	project, err := st.CreateProject("alice", "P")
	if err != nil {
		t.Fatal(err)
	}
	conv, err := st.CreateConversation("alice", string(project.ID), "Work", "")
	if err != nil {
		t.Fatal(err)
	}
	run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
	if err := st.CreateRun(&run); err != nil {
		t.Fatal(err)
	}
	doc := core.KnowledgeDoc{ID: "knowledge-1", Owner: "alice", ProjectID: project.ID, Source: "upload", Ref: "https://example.test/a", Chunk: "actual source"}
	if err := st.AddKnowledge(doc); err != nil {
		t.Fatal(err)
	}
	observation := store.RunWorkObservationInput{RunID: string(run.ID), SourceKind: "retrieval-hit", SourceID: string(doc.ID),
		SourceVersion: fmt.Sprintf("%x", sha256.Sum256([]byte(doc.Chunk))), Text: doc.Chunk, Delivery: "complete",
		Metadata: map[string]string{"ref": doc.Ref, "source": doc.Source}}
	if _, err := st.AppendRunWorkObservation(observation); err != nil {
		t.Fatal(err)
	}
	called := 0
	provider := workObserverForTest()
	provider.Observe = func(context.Context, extension.WorkObservation) (extension.WorkObservationResult, error) {
		called++
		return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: "rdm-1"}, nil
	}
	slot := &projectWorkObserverSlot{provider: &provider}
	work := func(context.Context, string, core.ID) (bool, error) { return true, nil }
	verify := func(ctx context.Context, row store.ProjectWorkObservationRow) error {
		return verifyProjectKnowledgeObservation(ctx, hostcaps{st: st}, row)
	}
	if n, err := deliverProjectWorkObservations(context.Background(), st, slot, work, 10, verify); err != nil || n != 1 || called != 1 {
		t.Fatalf("current owned source was not delivered: n=%d calls=%d err=%v", n, called, err)
	}
	if _, err := st.AppendRunWorkObservation(observation); err != nil {
		t.Fatal(err)
	}
	if _, err := st.DB().Exec(`DELETE FROM knowledge WHERE id=?`, doc.ID); err != nil {
		t.Fatal(err)
	}
	if n, err := deliverProjectWorkObservations(context.Background(), st, slot, work, 10, verify); err == nil || n != 0 || called != 1 {
		t.Fatalf("deleted source text crossed into observer: n=%d calls=%d err=%v", n, called, err)
	}
}

func TestTransientToolSourceKeepsCurrentRunAndCandidateStatus(t *testing.T) {
	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "tool.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()
	if err := st.EnsureUser("alice", "Alice"); err != nil {
		t.Fatal(err)
	}
	project, err := st.CreateProject("alice", "P")
	if err != nil {
		t.Fatal(err)
	}
	conv, err := st.CreateConversation("alice", string(project.ID), "Work", "")
	if err != nil {
		t.Fatal(err)
	}
	run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
	if err := st.CreateRun(&run); err != nil {
		t.Fatal(err)
	}
	_, err = st.AppendRunWorkObservation(store.RunWorkObservationInput{RunID: string(run.ID), SourceKind: "tool-result", SourceID: "call-1", SourceVersion: "v1",
		Text: "实际工具返回", Delivery: "complete", Metadata: map[string]string{"sourceStatus": "transient", "toolName": "lookup"}})
	if err != nil {
		t.Fatal(err)
	}
	rows, err := st.PendingProjectWorkObservations(10)
	if err != nil || len(rows) != 1 {
		t.Fatalf("tool source missing: %+v %v", rows, err)
	}
	if err := verifyProjectTransientObservation(context.Background(), st, rows[0]); err != nil {
		t.Fatalf("committed private tool source rejected: %v", err)
	}
	if _, err := st.DB().Exec(`DELETE FROM runs WHERE id=?`, run.ID); err != nil {
		t.Fatal(err)
	}
	if err := verifyProjectTransientObservation(context.Background(), st, rows[0]); err == nil {
		t.Fatal("deleted run still authorized its transient source")
	}
}

func TestRunPreflightWaitsForAllCommittedSourceParts(t *testing.T) {
	st, err := store.OpenSQLite(filepath.Join(t.TempDir(), "run-preflight.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer st.Close()
	if err := st.EnsureUser("alice", "Alice"); err != nil {
		t.Fatal(err)
	}
	project, err := st.CreateProject("alice", "P")
	if err != nil {
		t.Fatal(err)
	}
	conv, err := st.CreateConversation("alice", string(project.ID), "Work", "")
	if err != nil {
		t.Fatal(err)
	}
	run := core.Run{ConversationID: conv.ID, ProjectID: project.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunRunning, Phase: core.PhaseThinking}
	if err := st.CreateRun(&run); err != nil {
		t.Fatal(err)
	}
	msg := core.Message{ConversationID: conv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "检查成本和效率"}}}
	if err := st.AppendMessageWithRequestAccountChecked("alice", false, &msg, "req"); err != nil {
		t.Fatal(err)
	}
	if err := st.SetMessageRunIDAccountChecked("alice", string(msg.ID), string(run.ID), false); err != nil {
		t.Fatal(err)
	}
	if _, err := st.AppendRunWorkObservation(store.RunWorkObservationInput{RunID: string(run.ID), SourceKind: "tool-result", SourceID: "call-1",
		SourceVersion: "v1", Text: "工具中的测量结果", Delivery: "complete", Metadata: map[string]string{"sourceStatus": "transient", "toolName": "measure"}}); err != nil {
		t.Fatal(err)
	}
	ready, preflights, observed, verified := false, 0, 0, 0
	provider := workObserverForTest()
	provider.Observe = func(_ context.Context, observation extension.WorkObservation) (extension.WorkObservationResult, error) {
		observed++
		if observation.AuthorizationEpoch == "" {
			t.Fatal("observation omitted the current authorization epoch")
		}
		return extension.WorkObservationResult{State: extension.WorkObservationAccepted, ReceiptID: fmt.Sprintf("receipt-%d", observed)}, nil
	}
	provider.Preflight = func(_ context.Context, action extension.ProjectWorkAction) (extension.ProjectWorkDecision, error) {
		preflights++
		if action.RunID != run.ID || action.ProjectID != project.ID || len(action.Manifest) != 2 {
			t.Fatalf("run manifest omitted a committed source: %+v", action)
		}
		if !ready {
			return extension.ProjectWorkDecision{State: extension.ProjectWorkPending}, nil
		}
		return extension.ProjectWorkDecision{State: extension.ProjectWorkProceed, DecisionID: "approved-risk-check", SourceWatermark: action.SourceWatermark}, nil
	}
	slot := &projectWorkObserverSlot{provider: &provider}
	work := func(context.Context, string, core.ID) (bool, error) { return true, nil }
	verify := func(ctx context.Context, row store.ProjectWorkObservationRow) error {
		verified++
		return verifyProjectTransientObservation(ctx, st, row)
	}
	decision, err := preflightProjectRun(context.Background(), st, slot, work, verify, string(run.ID), "model.consume.tool")
	if err != nil || decision.State != extension.ProjectWorkPending || observed != 2 || preflights != 1 || verified != 2 {
		t.Fatalf("intake ACK was mistaken for finished risk analysis: %+v observed=%d preflights=%d err=%v", decision, observed, preflights, err)
	}
	ready = true
	decision, err = preflightProjectRun(context.Background(), st, slot, work, verify, string(run.ID), "model.consume.tool")
	if err != nil || decision.State != extension.ProjectWorkProceed || observed != 2 || preflights != 2 || verified != 3 {
		t.Fatalf("fully analyzed run was not released once: %+v observed=%d verified=%d preflights=%d err=%v", decision, observed, verified, preflights, err)
	}
}

func TestProjectWorkObservationRegistrationIsTrustedAndUnique(t *testing.T) {
	slot := &projectWorkObserverSlot{}
	trusted := pluginRegistry{pluginID: "rdm", caps: hostcaps{workObserver: slot}}
	other := pluginRegistry{pluginID: "other", caps: hostcaps{workObserver: slot}}
	if err := other.RegisterProjectWorkObserver(workObserverForTest()); err == nil {
		t.Fatal("untrusted plugin registered project observer")
	}
	if err := (compiledRegistry{}).RegisterProjectWorkObserver(workObserverForTest()); err == nil {
		t.Fatal("declarative plugin registered live observer callback")
	}
	if err := trusted.RegisterProjectWorkObserver(extension.ProjectWorkObserver{}); err == nil {
		t.Fatal("partial observer registered")
	}
	if err := trusted.RegisterProjectWorkObserver(workObserverForTest()); err != nil {
		t.Fatal(err)
	}
	if err := trusted.RegisterProjectWorkObserver(workObserverForTest()); err == nil {
		t.Fatal("second observer replaced the trusted provider")
	}
	if got := slot.get(); got == nil || got.Observe == nil || got.Preflight == nil || got.Respond == nil {
		t.Fatal("registered observer not available to host")
	}
}