From 18cd0d4247a6609673a6451007bb866a61a209e3 Mon Sep 17 00:00:00 2001 From: weichen Date: Wed, 14 Oct 2026 15:52:42 +0000 Subject: [PATCH] Persist private project message observations atomically Why: Project user messages and final assistant replies need exact, replayable source parts before a domain observer can safely analyze them. Evidence: go test -p 1 ./... passed; isolated PostgreSQL outbox, audit, migration, and work-context tests passed against a temporary cluster; request dedup, fenced settlement rollback, authorization recheck, and receipt idempotency tests passed. Impact: Add migration 44 and a private host outbox for committed message parts; actual observer delivery and action preflight are not yet wired into the product path. Refs: TT-N03, TT-R01; approved X3 stage-five plan r1 --- cmd/storage-migrate/main.go | 4 +- internal/app/project_work_observation.go | 58 ++++++ internal/app/project_work_observation_test.go | 52 +++++ internal/store/messages.go | 8 +- internal/store/migrate.go | 9 + internal/store/migrate_compat.go | 1 + internal/store/project_work_outbox.go | 181 ++++++++++++++++++ internal/store/project_work_outbox_test.go | 124 ++++++++++++ internal/store/recovery.go | 8 +- internal/store/run_account_write.go | 3 + 10 files changed, 444 insertions(+), 4 deletions(-) create mode 100644 internal/store/project_work_outbox.go create mode 100644 internal/store/project_work_outbox_test.go diff --git a/cmd/storage-migrate/main.go b/cmd/storage-migrate/main.go index 86d16db..eb395be 100644 --- a/cmd/storage-migrate/main.go +++ b/cmd/storage-migrate/main.go @@ -285,7 +285,7 @@ func cmdBackup(args []string) error { // admission_leases 是运行态执行许可（迁移无意义），明确排除。 var copyOrder = []string{ "projects", "users", "account_directory_revision", "profiles", "settings", "plugins", - "conversations", "runs", "messages", "events", "artifacts", + "conversations", "runs", "messages", "project_work_outbox", "events", "artifacts", "hitl_cards", "plans", "todos", "knowledge", "knowledge_vectors", "activity_cards", "project_event_shares", "command_audit", "project_audit_links", "prompt_modules", "sessions", "api_keys", @@ -953,7 +953,7 @@ func advanceSequences(db *sql.DB, table string) error { // 除外）。未知 owner 阻塞迁移（U4/R7：不静默带过未知归属）。 func checkOwners(db *sql.DB, src *sql.DB) error { ownerTables := []string{"projects", "conversations", "runs", "artifacts", "knowledge", "hitl_cards", "profiles", "activity_cards", - "project_event_shares", "command_audit", "project_audit_links", "work_task_bindings", "context_manifests", "workflow_step_sources", + "project_event_shares", "command_audit", "project_audit_links", "project_work_outbox", "work_task_bindings", "context_manifests", "workflow_step_sources", "sessions", "api_keys", "login_keys", "account_events", "recovery_proposals"} for _, t := range ownerTables { if !sqliteHasTable(src, t) { diff --git a/internal/app/project_work_observation.go b/internal/app/project_work_observation.go index 489ad1b..c6bca54 100644 --- a/internal/app/project_work_observation.go +++ b/internal/app/project_work_observation.go @@ -1,10 +1,14 @@ package app import ( + "context" "errors" + "fmt" "sync" + "agent-system-backend/internal/store" "agent-system-backend/pkg/extension" + core "agent-system-core" ) type projectWorkObserverSlot struct { @@ -54,3 +58,57 @@ func (r recordRegistry) RegisterProjectWorkObserver(p extension.ProjectWorkObser } return reg.RegisterProjectWorkObserver(p) } + +// deliverProjectWorkObservations replays only committed source parts, rechecking +// their owner and current project work authority before disclosing text to RDM. +// An Observe ACK is an intake receipt, not a preflight or publication decision. +// Failures leave the row pending for a later idempotent retry. +func deliverProjectWorkObservations(ctx context.Context, st *store.Store, slot *projectWorkObserverSlot, + work func(context.Context, string, core.ID) (bool, error), limit int, +) (int, error) { + provider := slot.get() + if provider == nil || work == nil { + return 0, errors.New("extension: trusted project work observer unavailable") + } + rows, err := st.PendingProjectWorkObservations(limit) + if err != nil { + return 0, err + } + completed := 0 + for _, row := range rows { + if err := ctx.Err(); err != nil { + return completed, err + } + conv, err := st.GetConversation(row.Owner, row.ConversationID) + if err != nil || string(conv.ProjectID) != row.ProjectID { + return completed, fmt.Errorf("project work source %s: current conversation binding unavailable", row.EventID) + } + allowed, err := work(ctx, row.Owner, core.ID(row.ProjectID)) + if err != nil || !allowed { + return completed, fmt.Errorf("project work source %s: current project authorization denied", row.EventID) + } + observation := extension.WorkObservation{ + EventID: row.EventID, Phase: "source-committed", Actor: row.Owner, Owner: row.Owner, + ProjectID: core.ID(row.ProjectID), ConversationID: core.ID(row.ConversationID), RunID: core.ID(row.RunID), + SourceKind: row.SourceKind, SourceID: row.SourceID, SourceVersion: row.SourceVersion, + SourceSHA256: row.SourceSHA256, PartIndex: row.PartIndex, SpanStart: row.SpanStart, SpanEnd: row.SpanEnd, + Text: row.Text, Scope: row.Scope, Delivery: row.Delivery, ObservedAt: row.ObservedAt, + } + result, err := provider.Observe(ctx, observation) + if err != nil { + return completed, err + } + switch result.State { + case extension.WorkObservationPending: + continue + case extension.WorkObservationAccepted, extension.WorkObservationPartial, extension.WorkObservationDenied: + if err := st.FinishProjectWorkObservation(row.EventID, result.ReceiptID, string(result.State)); err != nil { + return completed, err + } + completed++ + default: + return completed, fmt.Errorf("project work source %s: invalid observer result %q", row.EventID, result.State) + } + } + return completed, nil +} diff --git a/internal/app/project_work_observation_test.go b/internal/app/project_work_observation_test.go index 68b5ba7..079ab87 100644 --- a/internal/app/project_work_observation_test.go +++ b/internal/app/project_work_observation_test.go @@ -2,9 +2,12 @@ package app import ( "context" + "path/filepath" "testing" + "agent-system-backend/internal/store" "agent-system-backend/pkg/extension" + core "agent-system-core" ) func workObserverForTest() extension.ProjectWorkObserver { @@ -21,6 +24,55 @@ func workObserverForTest() extension.ProjectWorkObserver { } } +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 TestProjectWorkObservationRegistrationIsTrustedAndUnique(t *testing.T) { slot := &projectWorkObserverSlot{} trusted := pluginRegistry{pluginID: "rdm", caps: hostcaps{workObserver: slot}} diff --git a/internal/store/messages.go b/internal/store/messages.go index 061b24c..5cabdd7 100644 --- a/internal/store/messages.go +++ b/internal/store/messages.go @@ -48,6 +48,9 @@ func (s *Store) appendMessageWithRequest(m *core.Message, requestID, actor strin if m.ID == "" { m.ID = newID("msg") } + if m.CreatedAt.IsZero() { + m.CreatedAt = time.Now().UTC() + } m.CreatedAt = m.CreatedAt.UTC() insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` @@ -113,7 +116,10 @@ func (s *Store) appendMessageRows(tx *sql.Tx, m *core.Message, requestID, insert if err := s.dialect.lockConversation(tx, string(m.ConversationID)); err != nil { return err } - if _, err := tx.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), nowISO(), requestID, m.ConversationID); err != nil { + if _, err := tx.Exec(s.q(insert), m.ID, m.ConversationID, m.RunID, string(m.Role), marshalJSON(m.Parts), m.CreatedAt.Format(time.RFC3339Nano), requestID, m.ConversationID); err != nil { + return err + } + if err := s.appendProjectMessageObservationsTx(tx, m); err != nil { return err } _, err := tx.Exec(s.q(`UPDATE conversations SET updated_at=? WHERE id=?`), nowISO(), m.ConversationID) diff --git a/internal/store/migrate.go b/internal/store/migrate.go index 57e43a1..a773e0e 100644 --- a/internal/store/migrate.go +++ b/internal/store/migrate.go @@ -1085,6 +1085,15 @@ func init() { `DROP FUNCTION project_audit_writers_guarded(regclass)`, }, }}, + migration{version: 44, name: "project_work_observation_outbox", + up: map[string][]string{ + "sqlite": {`CREATE TABLE project_work_outbox (event_id TEXT PRIMARY KEY,owner TEXT NOT NULL,project_id TEXT NOT NULL,conversation_id TEXT NOT NULL,run_id TEXT NOT NULL DEFAULT '',source_kind TEXT NOT NULL,source_id TEXT NOT NULL,source_version TEXT NOT NULL,source_sha256 TEXT NOT NULL,part_index INTEGER NOT NULL,span_start INTEGER NOT NULL,span_end INTEGER NOT NULL,delivery TEXT NOT NULL,scope TEXT NOT NULL,observed_at TEXT NOT NULL,state TEXT NOT NULL DEFAULT 'pending',receipt_id TEXT NOT NULL DEFAULT '',UNIQUE(source_kind,source_id,part_index))`, `CREATE INDEX idx_project_work_outbox_pending ON project_work_outbox(state,observed_at,event_id)`}, + "postgres": {`CREATE TABLE project_work_outbox (event_id TEXT PRIMARY KEY,owner TEXT NOT NULL,project_id TEXT NOT NULL,conversation_id TEXT NOT NULL,run_id TEXT NOT NULL DEFAULT '',source_kind TEXT NOT NULL,source_id TEXT NOT NULL,source_version TEXT NOT NULL,source_sha256 TEXT NOT NULL,part_index INTEGER NOT NULL,span_start INTEGER NOT NULL,span_end INTEGER NOT NULL,delivery TEXT NOT NULL,scope TEXT NOT NULL,observed_at TEXT NOT NULL,state TEXT NOT NULL DEFAULT 'pending',receipt_id TEXT NOT NULL DEFAULT '',UNIQUE(source_kind,source_id,part_index))`, `CREATE INDEX idx_project_work_outbox_pending ON project_work_outbox(state,observed_at,event_id)`}, + }, + down: map[string][]string{ + "sqlite": {`DROP INDEX IF EXISTS idx_project_work_outbox_pending`, `DROP TABLE IF EXISTS project_work_outbox`}, + "postgres": {`DROP INDEX IF EXISTS idx_project_work_outbox_pending`, `DROP TABLE IF EXISTS project_work_outbox`}, + }}, ) } diff --git a/internal/store/migrate_compat.go b/internal/store/migrate_compat.go index b7623d5..97d8efd 100644 --- a/internal/store/migrate_compat.go +++ b/internal/store/migrate_compat.go @@ -48,6 +48,7 @@ var migrationBreaksOldReaders = map[int]bool{ 41: false, // project_audit_committed_links：新表，旧读者不受影响 42: false, // project_audit_committed_prefix：PG 审计 ID 默认函数，旧读者兼容 43: false, // project_audit_all_writers_guard：PG 在线审计写入门禁，旧读者兼容 + 44: false, // project_work_observation_outbox：独立私域投递表，旧读者不读取 } // SchemaTarget 是本二进制会迁移到的 schema 版本。 diff --git a/internal/store/project_work_outbox.go b/internal/store/project_work_outbox.go new file mode 100644 index 0000000..9f39c26 --- /dev/null +++ b/internal/store/project_work_outbox.go @@ -0,0 +1,181 @@ +package store + +import ( + "crypto/sha256" + "database/sql" + "encoding/hex" + "errors" + "fmt" + "time" + + core "agent-system-core" +) + +// ProjectWorkObservationRow is a private host delivery record. A project ID is +// a trusted locator here, not authorization to reveal the original message to +// other project members. The source body is fetched from the message row only +// for the current provider delivery and is never part of project audit DTOs. +type ProjectWorkObservationRow struct { + EventID string + Owner string + ProjectID string + ConversationID string + RunID string + SourceKind string + SourceID string + SourceVersion string + SourceSHA256 string + PartIndex int + SpanStart int + SpanEnd int + Delivery string + Scope string + ObservedAt time.Time + State string + ReceiptID string + Text string + Reason string +} + +func (s *Store) appendProjectMessageObservationsTx(tx *sql.Tx, m *core.Message) error { + if m == nil || (m.Role != core.RoleUser && m.Role != core.RoleAssistant) { + return nil + } + var owner, projectID string + if err := tx.QueryRow(s.q(`SELECT owner,project_id FROM conversations WHERE id=?`), m.ConversationID).Scan(&owner, &projectID); err != nil { + return err + } + if projectID == "" { + return nil + } + kind := string(m.Role) + "-message" + observed := m.CreatedAt.UTC().Format(time.RFC3339Nano) + for i, part := range m.Parts { + // A tool summary, artifact reference, card, or hidden thinking part is not + // evidence of the original resource body. The separate retrieval/tool + // adapters will supply their actual source IDs and spans later. + if part.Type == core.PartThinking { + continue + } + delivery, text := "partial", "" + var material []byte + if part.Type == core.PartText && part.Text != "" { + delivery, text = "complete", part.Text + material = []byte(text) + } else { + material = []byte(marshalJSON(part)) + } + hash := sha256.Sum256(material) + idDigest := sha256.Sum256([]byte(fmt.Sprintf("%s:%s:%d", kind, m.ID, i))) + eventID := "obs_" + hex.EncodeToString(idDigest[:16]) + if _, err := tx.Exec(s.q(`INSERT INTO project_work_outbox(event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`), eventID, owner, projectID, m.ConversationID, m.RunID, + kind, m.ID, m.ID, hex.EncodeToString(hash[:]), i, 0, len(text), delivery, "private", observed); err != nil { + return err + } + } + return nil +} + +// PendingProjectWorkObservations returns committed, not-yet-ACKed source parts. +// It rechecks that the immutable source bytes still match the recorded hash. +// Missing or changed source bytes become partial, never invented complete text. +func (s *Store) PendingProjectWorkObservations(limit int) ([]ProjectWorkObservationRow, error) { + if limit <= 0 || limit > 200 { + limit = 50 + } + rows, err := s.db.Query(s.q(`SELECT event_id,owner,project_id,conversation_id,run_id,source_kind,source_id,source_version,source_sha256,part_index,span_start,span_end,delivery,scope,observed_at,state,receipt_id + FROM project_work_outbox WHERE state='pending' ORDER BY observed_at,event_id LIMIT ?`), limit) + if err != nil { + return nil, err + } + out := make([]ProjectWorkObservationRow, 0, limit) + for rows.Next() { + var row ProjectWorkObservationRow + var observed string + if err := rows.Scan(&row.EventID, &row.Owner, &row.ProjectID, &row.ConversationID, &row.RunID, + &row.SourceKind, &row.SourceID, &row.SourceVersion, &row.SourceSHA256, &row.PartIndex, + &row.SpanStart, &row.SpanEnd, &row.Delivery, &row.Scope, &observed, &row.State, &row.ReceiptID); err != nil { + rows.Close() + return nil, err + } + row.ObservedAt, err = time.Parse(time.RFC3339Nano, observed) + if err != nil { + rows.Close() + return nil, err + } + out = append(out, row) + } + err = rows.Err() + rows.Close() + if err != nil { + return nil, err + } + for i := range out { + row := &out[i] + if row.Delivery != "complete" { + continue + } + msg, err := s.MessageByID(row.SourceID) + if errors.Is(err, ErrNotFound) { + row.Delivery, row.Reason = "partial", "source-missing" + continue + } + if err != nil { + return nil, err + } + if msg.ConversationID != core.ID(row.ConversationID) || row.PartIndex < 0 || row.PartIndex >= len(msg.Parts) || msg.Parts[row.PartIndex].Type != core.PartText { + row.Delivery, row.Reason = "partial", "source-changed" + continue + } + text := msg.Parts[row.PartIndex].Text + hash := sha256.Sum256([]byte(text)) + if hex.EncodeToString(hash[:]) != row.SourceSHA256 || row.SpanEnd != len(text) { + row.Delivery, row.Reason = "partial", "source-changed" + continue + } + row.Text = text + } + return out, nil +} + +// AckProjectWorkObservation records only the provider's durable intake receipt. +// It does not assert that source analysis, risk checks, or project publication +// finished. A different receipt for the same event is a conflict, not a retry. +func (s *Store) AckProjectWorkObservation(eventID, receiptID string) error { + return s.FinishProjectWorkObservation(eventID, receiptID, "accepted") +} + +// FinishProjectWorkObservation records the provider's terminal intake result. +// "accepted" is only durable receipt; "partial" and "denied" are truthful +// terminal treatment of this source part, not a project fact publication. +func (s *Store) FinishProjectWorkObservation(eventID, receiptID, state string) error { + if eventID == "" || receiptID == "" { + return errors.New("project work observation: empty event or receipt") + } + if state != "accepted" && state != "partial" && state != "denied" { + return errors.New("project work observation: invalid terminal state") + } + res, err := s.db.Exec(s.q(`UPDATE project_work_outbox SET state=?,receipt_id=? WHERE event_id=? AND state='pending'`), state, receiptID, eventID) + if err != nil { + return err + } + changed, err := res.RowsAffected() + if err != nil { + return err + } + if changed == 1 { + return nil + } + var currentReceipt, currentState string + if err := s.db.QueryRow(s.q(`SELECT receipt_id,state FROM project_work_outbox WHERE event_id=?`), eventID).Scan(¤tReceipt, ¤tState); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return ErrNotFound + } + return err + } + if currentReceipt != receiptID || currentState != state { + return fmt.Errorf("project work observation: receipt conflict for %s", eventID) + } + return nil +} diff --git a/internal/store/project_work_outbox_test.go b/internal/store/project_work_outbox_test.go new file mode 100644 index 0000000..38c9b57 --- /dev/null +++ b/internal/store/project_work_outbox_test.go @@ -0,0 +1,124 @@ +package store + +import ( + "crypto/sha256" + "fmt" + "testing" + "time" + + core "agent-system-core" +) + +func TestProjectMessageOutboxFollowsCommittedMessage(t *testing.T) { + st := openTestStore(t) + if err := st.EnsureUser("alice", "Alice"); err != nil { + t.Fatal(err) + } + p, err := st.CreateProject("alice", "Project") + if err != nil { + t.Fatal(err) + } + projectConv, err := st.CreateConversation("alice", string(p.ID), "Work", "") + if err != nil { + t.Fatal(err) + } + privateConv, err := st.CreateConversation("alice", string(p.ID), "Private", "") + if err != nil { + t.Fatal(err) + } + msg := core.Message{ConversationID: projectConv.ID, Role: core.RoleUser, Parts: []core.Part{ + {Type: core.PartText, Text: "要求效率和成本都低于人工"}, + {Type: core.PartArtifactRef, Artifact: &core.ArtifactRef{ArtifactID: "file-1"}}, + }} + if err := st.AppendMessageWithRequestAccountChecked("alice", false, &msg, "request-1"); err != nil { + t.Fatal(err) + } + rows, err := st.PendingProjectWorkObservations(10) + if err != nil { + t.Fatal(err) + } + if len(rows) != 2 { + t.Fatalf("message and artifact parts were not each recorded: %+v", rows) + } + byPart := map[int]ProjectWorkObservationRow{} + for _, row := range rows { + byPart[row.PartIndex] = row + } + wantHash := fmt.Sprintf("%x", sha256.Sum256([]byte(msg.Parts[0].Text))) + textPart := byPart[0] + if textPart.ProjectID != string(p.ID) || textPart.Owner != "alice" || textPart.SourceID != string(msg.ID) || textPart.SourceSHA256 != wantHash || textPart.Text != msg.Parts[0].Text || textPart.Delivery != "complete" || textPart.RunID != "" { + t.Fatalf("incorrect trusted user message observation: %+v", textPart) + } + artifactPart := byPart[1] + if artifactPart.Delivery != "partial" || artifactPart.Text != "" || artifactPart.SourceID != string(msg.ID) { + t.Fatalf("unparsed attachment was presented as analyzed text: %+v", artifactPart) + } + run := core.Run{ConversationID: projectConv.ID, ProjectID: p.ID, Owner: "alice", Kind: core.RunConversation, Status: core.RunPending, Phase: core.PhaseThinking} + if err := st.CreateRun(&run); err != nil { + t.Fatal(err) + } + if err := st.SetMessageRunIDAccountChecked("alice", string(msg.ID), string(run.ID), false); err != nil { + t.Fatal(err) + } + linked, err := st.PendingProjectWorkObservations(10) + if err != nil || len(linked) != 2 || linked[0].RunID != string(run.ID) || linked[1].RunID != string(run.ID) { + t.Fatalf("later trusted run link was not saved for every source part: %+v %v", linked, err) + } + retry := core.Message{ConversationID: projectConv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "changed"}}} + if err := st.AppendMessageWithRequestAccountChecked("alice", false, &retry, "request-1"); err == nil { + t.Fatal("duplicate request did not surface existing message") + } + private := core.Message{ConversationID: privateConv.ID, Role: core.RoleUser, Parts: []core.Part{{Type: core.PartText, Text: "private"}}} + if err := st.AppendMessageWithRequestAccountChecked("alice", false, &private, "request-2"); err != nil { + t.Fatal(err) + } + again, err := st.PendingProjectWorkObservations(10) + if err != nil || len(again) != 3 { + t.Fatalf("duplicate request generated an extra source or private source was lost: %+v %v", again, err) + } + for _, row := range again { + if row.Scope != "private" { + t.Fatalf("project locator alone changed a message's private scope: %+v", row) + } + } + if err := st.AckProjectWorkObservation(textPart.EventID, "receipt-1"); err != nil { + t.Fatal(err) + } + if err := st.AckProjectWorkObservation(textPart.EventID, "receipt-1"); err != nil { + t.Fatalf("same committed receipt was not idempotent: %v", err) + } + if err := st.AckProjectWorkObservation(textPart.EventID, "different"); err == nil { + t.Fatal("different observer receipt replaced the first one") + } + pending, err := st.PendingProjectWorkObservations(10) + if err != nil || len(pending) != 2 { + t.Fatalf("ACK did not remove only the delivered source: %+v %v", pending, err) + } +} + +func TestAssistantObservationCommitsOnlyWithFencedRunSettlement(t *testing.T) { + st := openTestStore(t) + conv := seedConversation(t, st) + run := seedRun(t, st, conv, core.RunRunning) + now := time.Now().UTC() + run.Status, run.Phase, run.EndedAt = core.RunCompleted, core.PhaseDone, &now + answer := core.Message{ConversationID: conv.ID, RunID: run.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "需要澄清效率的含义"}}} + updated, err := st.SettleRun(&run, &answer) + if err != nil || !updated { + t.Fatalf("run settlement failed: updated=%t err=%v", updated, err) + } + rows, err := st.PendingProjectWorkObservations(10) + if err != nil || len(rows) != 1 || rows[0].SourceKind != "assistant-message" || rows[0].SourceID != string(answer.ID) || rows[0].RunID != string(run.ID) || rows[0].Text != answer.Parts[0].Text { + t.Fatalf("final assistant observation was not committed with the run: %+v %v", rows, err) + } + secondRun := seedRun(t, st, conv, core.RunRunning) + secondRun.Status, secondRun.Phase, secondRun.EndedAt = core.RunCompleted, core.PhaseDone, &now + failed := core.Message{ID: answer.ID, ConversationID: conv.ID, RunID: secondRun.ID, Role: core.RoleAssistant, Parts: []core.Part{{Type: core.PartText, Text: "different"}}} + if _, err := st.SettleRun(&secondRun, &failed); err == nil { + t.Fatal("duplicate final message did not fail") + } + again, err := st.PendingProjectWorkObservations(10) + if err != nil || len(again) != 1 { + t.Fatalf("failed settlement added a second observation: %+v %v", again, err) + } +} diff --git a/internal/store/recovery.go b/internal/store/recovery.go index 6adfb0b..3674046 100644 --- a/internal/store/recovery.go +++ b/internal/store/recovery.go @@ -256,9 +256,15 @@ func (s *Store) settleRunOnce(r *core.Run, msg *core.Message, fence *fenceRef, a return false, false, err } if msg != nil { + if msg.CreatedAt.IsZero() { + msg.CreatedAt = time.Now().UTC() + } insert := `INSERT INTO messages(id,conversation_id,run_id,role,parts,seq,created_at,request_id) SELECT ?,?,?,?,?,COALESCE(MAX(seq),0)+1,?,? FROM messages WHERE conversation_id=?` - if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), nowISO(), "", msg.ConversationID); err != nil { + if _, err := tx.Exec(s.q(insert), msg.ID, msg.ConversationID, msg.RunID, string(msg.Role), marshalJSON(msg.Parts), msg.CreatedAt.UTC().Format(time.RFC3339Nano), "", msg.ConversationID); err != nil { + return false, false, err + } + if err := s.appendProjectMessageObservationsTx(tx, msg); err != nil { return false, false, err } if msg.VariantGroup != "" || msg.Excluded { diff --git a/internal/store/run_account_write.go b/internal/store/run_account_write.go index 2cefab7..636b086 100644 --- a/internal/store/run_account_write.go +++ b/internal/store/run_account_write.go @@ -96,6 +96,9 @@ func (s *Store) SetMessageRunIDAccountChecked(actor, messageID, runID string, al } else if n != 1 { return ErrNotFound } + if _, err := tx.Exec(s.q(`UPDATE project_work_outbox SET run_id=? WHERE source_kind='user-message' AND source_id=?`), runID, messageID); err != nil { + return err + } return nil }) }