Agent-System-Backend/eval/live/work_checkpoints_test.go

//go:build live

package live

import (
	core "agent-system-core"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"syscall"
	"testing"
	"time"

	_ "modernc.org/sqlite"
)

type checkpointManifest struct {
	PID                                                           int `json:"pid"`
	Binary, BinarySHA256, BackendSHA, DevSHA, Base, DataDir, Mode string
}

func checkpointEvalOutputName(filter, authorization string) (string, bool) {
	if filter == "A01" && authorization == "approved-20261013" {
		return "candidate-extra-a01", true
	}
	if filter == "A01" && authorization == "approved-followup-20261013" {
		return "candidate-followup-a01", true
	}
	if filter == "A01" && authorization == "approved-r02-20261013" {
		return "candidate-r02-a01", true
	}
	if filter == "A01" && authorization == "approved-r03-20261013" {
		return "candidate-r03-a01", true
	}
	if filter == "A01" && authorization == "approved-r04-20261014" {
		return "candidate-r04-a01", true
	}
	if filter == "A02,A04" {
		return "candidate-remaining", true
	}
	return "", false
}

func persistedRunningCheckpointMax(path, conversationID string) (int, error) {
	db, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		return 0, err
	}
	defer db.Close()
	rows, err := db.Query(`SELECT payload FROM events WHERE conversation_id = ? AND type = 'card.updated' ORDER BY seq`, conversationID)
	if err != nil {
		return 0, err
	}
	defer rows.Close()
	maximum := 0
	for rows.Next() {
		var raw []byte
		if err = rows.Scan(&raw); err != nil {
			return 0, err
		}
		var event struct {
			Card struct {
				Status string `json:"status"`
				Task   struct {
					ExecutionState string `json:"executionState"`
					CheckpointSeq  int    `json:"checkpointSeq"`
				} `json:"task"`
			} `json:"card"`
		}
		if err = json.Unmarshal(raw, &event); err != nil {
			return 0, err
		}
		if event.Card.Status == "active" && event.Card.Task.ExecutionState == "running" && event.Card.Task.CheckpointSeq > maximum {
			maximum = event.Card.Task.CheckpointSeq
		}
	}
	return maximum, rows.Err()
}

func TestFollowupA01KeepsItsOwnOneShotResult(t *testing.T) {
	for _, tc := range []struct {
		filter, authorization, name string
		allowed                     bool
	}{
		{"A01", "approved-20261013", "candidate-extra-a01", true},
		{"A01", "approved-followup-20261013", "candidate-followup-a01", true},
		{"A01", "approved-r02-20261013", "candidate-r02-a01", true},
		{"A01", "approved-r03-20261013", "candidate-r03-a01", true},
		{"A01", "approved-r04-20261014", "candidate-r04-a01", true},
		{"A02,A04", "approved-r03-20261013", "candidate-remaining", true},
		{"A01", "", "", false},
		{"A01", "unknown", "", false},
		{"A02,A04", "", "candidate-remaining", true},
	} {
		name, allowed := checkpointEvalOutputName(tc.filter, tc.authorization)
		if name != tc.name || allowed != tc.allowed {
			t.Fatalf("filter=%q authorization=%q: got %q/%v; want %q/%v", tc.filter, tc.authorization, name, allowed, tc.name, tc.allowed)
		}
	}
}

func verifiedCheckpointManifest(t *testing.T) checkpointManifest {
	t.Helper()
	path := os.Getenv("WORK_CHECKPOINT_EVAL_OWNERSHIP_MANIFEST")
	if path == "" {
		t.Fatal("explicit task-owned process manifest required")
	}
	raw, e := os.ReadFile(path)
	if e != nil {
		t.Fatal(e)
	}
	var m checkpointManifest
	if json.Unmarshal(raw, &m) != nil {
		t.Fatal("invalid manifest")
	}
	u, e := url.Parse(m.Base)
	if e != nil || u.Hostname() != "127.0.0.1" || m.Base != os.Getenv("AGENT_EVAL_BASE") || m.PID <= 0 || !filepath.IsAbs(m.DataDir) || !strings.Contains(m.DataDir, "fast-slow-u2-u3-") {
		t.Fatal("manifest scope mismatch")
	}
	if e = syscall.Kill(m.PID, 0); e != nil {
		t.Fatal("owned process is not live")
	}
	actual, e := os.Readlink(fmt.Sprintf("/proc/%d/exe", m.PID))
	if e != nil || actual != m.Binary {
		t.Fatal("process executable mismatch")
	}
	bytes, e := os.ReadFile(actual)
	if e != nil {
		t.Fatal(e)
	}
	sum := sha256.Sum256(bytes)
	if hex.EncodeToString(sum[:]) != m.BinarySHA256 {
		t.Fatal("binary hash mismatch")
	}
	if _, e = os.Stat(filepath.Join(m.DataDir, "agent-system.db")); e != nil {
		t.Fatal("isolated data missing")
	}
	return m
}

// The batch driver invokes this once per baseline/candidate, using the same
// three synthetic materials, route and profile. It never resolves a business HITL.
func TestWorkCheckpoints(t *testing.T) {
	m := verifiedCheckpointManifest(t)
	if m.Mode != "model" {
		t.Fatal("real model evaluation requires model mode")
	}
	materials := os.Getenv("AGENT_EVAL_MATERIALS")
	outDir := os.Getenv("AGENT_EVAL_OUT")
	if materials == "" || outDir == "" {
		t.Fatal("explicit external material and evidence directories required")
	}
	if e := os.MkdirAll(outDir, 0700); e != nil {
		t.Fatal(e)
	}
	c := &client{base: m.Base, http: &http.Client{Timeout: 30 * time.Second}, t: t}
	label := os.Getenv("WORK_CHECKPOINT_EVAL_LABEL")
	if label != "baseline" && label != "candidate" {
		t.Fatal("fixed baseline/candidate label required")
	}
	caseFilter := os.Getenv("WORK_CHECKPOINT_EVAL_CASES")
	outputName := label
	if caseFilter != "" {
		if label != "candidate" {
			t.Fatal("candidate cases only")
		}
		var allowed bool
		outputName, allowed = checkpointEvalOutputName(caseFilter, os.Getenv("WORK_CHECKPOINT_EXTRA_A01"))
		if !allowed {
			t.Fatal("explicitly authorized case selection required")
		}
	}
	if _, err := os.Stat(filepath.Join(outDir, outputName+".json")); err == nil {
		t.Fatal("sample already exists; do not retry")
	}
	all := []map[string]any{}
	total := 0
	for _, testCase := range []struct {
		id, focus string
		files     []string
	}{{"A01", "", []string{"需求.txt", "会议.txt", "评审.txt"}}, {"A02", "重点核对预算、外部依赖和交付日期；请保留限制", []string{"需求.txt", "会议.txt", "评审.txt"}}, {"A04", "", []string{"需求.txt", "会议.txt", "扫描.pdf"}}} {
		if caseFilter != "" && !strings.Contains(caseFilter, testCase.id) {
			continue
		}
		if total >= 120000 {
			t.Fatal("instance reported token budget exhausted")
		}
		var project struct {
			ID string `json:"id"`
		}
		if e := c.do("POST", "/api/projects", map[string]any{"name": "checkpoint-" + label + "-" + testCase.id}, &project); e != nil {
			t.Fatal(e)
		}
		var conv struct {
			ID string `json:"id"`
		}
		if e := c.do("POST", "/api/projects/"+project.ID+"/conversations", map[string]any{"title": testCase.id, "profileId": "checkpoint-eval"}, &conv); e != nil {
			t.Fatal(e)
		}
		ids := []string{}
		for _, name := range testCase.files {
			bytes, e := os.ReadFile(filepath.Join(materials, name))
			if e != nil {
				t.Fatal(e)
			}
			ids = append(ids, c.upload(t, project.ID, name, bytes))
		}
		var send map[string]any
		if e := c.do("POST", "/api/conversations/"+conv.ID+"/messages", map[string]any{"text": testCase.focus, "attachmentIds": ids, "requestId": "checkpoint-" + label + "-" + testCase.id}, &send); e != nil {
			t.Fatal(e)
		}
		deadline := time.Now().Add(300 * time.Second)
		var snapshot struct {
			Cards     []core.ActivityCard `json:"cards"`
			ReplaySeq int64               `json:"replaySeq"`
		}
		observations := []map[string]any{}
		seen := map[string]bool{}
		for {
			// Decode into fresh storage: reusing a slice lets JSON overwrite Task
			// pointers already saved as historical observations.
			snapshot.Cards = nil
			if e := c.do("GET", "/api/conversations/"+conv.ID+"/cards", nil, &snapshot); e != nil {
				t.Fatal(e)
			}
			terminal := false
			for _, card := range snapshot.Cards {
				if card.Kind != core.CardKindProgress {
					continue
				}
				if card.Task != nil {
					key := fmt.Sprintf("%s:%d", card.ID, card.Task.Revision)
					if !seen[key] {
						seen[key] = true
						observations = append(observations, map[string]any{"observedAt": time.Now().UTC(), "card": card, "replaySeq": snapshot.ReplaySeq})
					}
					terminal = card.Task.Terminal()
				} else {
					terminal = card.Status == core.CardDone || card.Status == core.CardFailed
				}
			}
			if terminal {
				break
			}
			if label == "baseline" && time.Until(deadline) < 298*time.Second && len(snapshot.Cards) == 0 {
				runObj, ok := send["run"].(map[string]any)
				if ok {
					var state struct {
						Run core.Run `json:"run"`
					}
					if c.do("GET", "/api/runs/"+fmt.Sprint(runObj["id"]), nil, &state) == nil && state.Run.Status == core.RunCompleted {
						break
					}
				}
			}
			if time.Now().After(deadline) {
				t.Fatalf("%s background did not finish within budget", testCase.id)
			}
			time.Sleep(250 * time.Millisecond)
		}
		var detail map[string]json.RawMessage
		if e := c.do("GET", "/api/conversations/"+conv.ID, nil, &detail); e != nil {
			t.Fatal(e)
		}
		var runs []core.Run
		_ = json.Unmarshal(detail["runs"], &runs)
		for _, r := range runs {
			if r.Usage != nil {
				total += r.Usage.TotalTokens
			}
		}
		record := map[string]any{"case": testCase.id, "projectId": project.ID, "conversationId": conv.ID, "send": send, "observations": observations, "finalCards": snapshot.Cards, "detail": detail, "reportedTokensCumulative": total, "quality": "requires independent comparison against provided materials; quote existence is not semantic proof"}
		all = append(all, record)
		raw, _ := json.MarshalIndent(map[string]any{"manifest": m, "label": label, "cases": all, "reportedTokens": total, "noticeModelTurns": "disabled equally in controlled comparison"}, "", "  ")
		if e := os.WriteFile(filepath.Join(outDir, outputName+".json"), raw, 0600); e != nil {
			t.Fatal(e)
		}
		if label == "candidate" && testCase.id == "A01" {
			maximum, err := persistedRunningCheckpointMax(filepath.Join(m.DataDir, "agent-system.db"), conv.ID)
			if err != nil {
				t.Fatal(err)
			}
			if maximum < 2 {
				t.Fatal("A01 has fewer than two persisted running checkpoints; inspect raw evidence")
			}
		}
		if total > 120000 {
			t.Fatal("reported instance token target exceeded; no further sample started")
		}
	}
	t.Logf("%s: %d material tasks, cumulative reported tokens=%d; raw quality evidence=%s", label, len(all), total, filepath.Join(outDir, outputName+".json"))
}