diff --git a/cmd/document-revision-backfill/main.go b/cmd/document-revision-backfill/main.go index 3698deb3..3bc74bba 100644 --- a/cmd/document-revision-backfill/main.go +++ b/cmd/document-revision-backfill/main.go @@ -60,7 +60,7 @@ func main() { Mismatch int `json:"hash_mismatch"` }{DryRun: !*apply} - svc := docrevision.NewService(nil) + svc := docrevision.NewService(pool) for rows.Next() { var docID, workDir, dbSHA, state string var projectID int64 diff --git a/internal/docrevision/bootstrap_test.go b/internal/docrevision/bootstrap_test.go new file mode 100644 index 00000000..3a6d782d --- /dev/null +++ b/internal/docrevision/bootstrap_test.go @@ -0,0 +1,94 @@ +package docrevision_test + +import ( + "context" + "testing" + + "chat-agent-system/internal/docrevision" +) + +func TestEnsureBootstrapIdempotentAndReadable(t *testing.T) { + svc := docrevision.NewService(nil) + ctx := context.Background() + st, err := svc.EnsureBootstrap(ctx, docrevision.BootstrapRequest{ + DocID: "doc_proj143_srs", Kind: "controlled", Markdown: "# SRS\n\nbody\n", + ProjectID: 143, WorkDir: "workspaces/p143/srs", LifecycleState: "drafting", + Actor: docrevision.ActorSnapshot{DisplayName: "bootstrap"}, + }) + if err != nil { + t.Fatal(err) + } + if st.CurrentVersionNo != 1 || st.HeadRevisionNo != 0 { + t.Fatalf("unexpected state %+v", st) + } + st2, err := svc.EnsureBootstrap(ctx, docrevision.BootstrapRequest{ + DocID: "doc_proj143_srs", Markdown: "# other\n", ProjectID: 143, + }) + if err != nil { + t.Fatal(err) + } + if st2.HeadSHA256 != st.HeadSHA256 { + t.Fatalf("bootstrap should be idempotent") + } + got, err := svc.GetRevisionState(ctx, "doc_proj143_srs") + if err != nil || got.ETag == "" { + t.Fatalf("get after bootstrap: %v %+v", err, got) + } +} + +func TestEnsureBootstrapNeedsContent(t *testing.T) { + svc := docrevision.NewService(nil) + _, err := svc.EnsureBootstrap(context.Background(), docrevision.BootstrapRequest{DocID: "doc-upload-x"}) + if err != docrevision.ErrNeedsContent { + t.Fatalf("want ErrNeedsContent got %v", err) + } +} + +func TestDeterministicChangeReportNoStaticFiction(t *testing.T) { + before := docrevision.CanonicalizeMarkdown("# A\n\n旧文\n") + after := docrevision.CanonicalizeMarkdown("# A\n\n新文 REQ-1\n") + fb := docrevision.BuildBlockMap(before, nil) + ab := docrevision.BuildBlockMap(after, fb) + diff := docrevision.ComputeDiff(before, after, fb, ab) + from := docrevision.SHA256Hex(before) + to := docrevision.SHA256Hex(after) + raw := docrevision.BuildDeterministicChangeReportJSON(from, to, docrevision.RiskHigh, nil, diff) + if raw == "" { + t.Fatal("empty report") + } + rep, err := docrevision.ValidateChangeReportJSON(raw, from, to, docrevision.RiskHigh, nil, diff.Hunks) + if err != nil { + t.Fatal(err) + } + if len(rep.Summary) == 0 { + t.Fatal("expected deterministic summary") + } + for _, s := range rep.Summary { + if s == "智能体交付物已生成" || s == "系统需求规格已更新" { + t.Fatalf("static fiction summary leaked: %q", s) + } + } +} + +func TestHumanSaveAfterBootstrap(t *testing.T) { + svc := docrevision.NewService(nil) + ctx := context.Background() + st, _ := svc.EnsureBootstrap(ctx, docrevision.BootstrapRequest{ + DocID: "doc_edit", Markdown: "# A\n", ProjectID: 1, LifecycleState: "drafting", + }) + r, err := svc.CommitHumanRevision(ctx, docrevision.HumanRevisionRequest{ + DocID: "doc_edit", Markdown: "# A\n\nedit\n", + ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo, + IdempotencyKey: "e1", ActorUserID: 9, LifecycleState: "drafting", + }) + if err != nil { + t.Fatal(err) + } + if r.State.HeadRevisionNo != 1 { + t.Fatalf("want r1 got %d", r.State.HeadRevisionNo) + } + ov, err := svc.ManualOverlay(ctx, "doc_edit") + if err != nil || ov == nil || (ov.Summary.Modifies+ov.Summary.Inserts) == 0 { + t.Fatalf("manual overlay empty: %+v %v", ov, err) + } +} diff --git a/internal/docrevision/pg_mutate.go b/internal/docrevision/pg_mutate.go new file mode 100644 index 00000000..09eda726 --- /dev/null +++ b/internal/docrevision/pg_mutate.go @@ -0,0 +1,224 @@ +package docrevision + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5" +) + +func (s *Service) openTemplatePGLocked(ctx context.Context, docID, canon, sha string, actor ActorSnapshot, projectID int64, workDir string) (*RevisionState, error) { + blob := &ContentBlob{ContentSHA256: sha, ContentMarkdown: canon, ByteSize: int64(len(canon)), CreatedAt: time.Now().UTC()} + blocks := BuildBlockMap(canon, nil) + err := s.withTx(ctx, func(tx pgx.Tx) error { + if err := s.ensureBlobTX(ctx, tx, blob); err != nil { + return err + } + sn := &Snapshot{ + DocID: docID, ContentSHA256: sha, BlockMap: blocks, SourceKind: SourceTemplate, + ActorRef: "template", ActorSnapshot: actor, CreatedAt: time.Now().UTC(), + } + id, err := s.insertSnapshotTX(ctx, tx, sn) + if err != nil { + return err + } + sn.ID = id + s.snapshots[id] = sn + + ver := &Version{ + DocID: docID, VersionNo: 1, Status: VersionOpen, OriginKind: OriginTemplate, + BaselineSnapshotID: id, HeadSnapshotID: id, OpenedAt: time.Now().UTC(), + } + vid, err := s.insertVersionTX(ctx, tx, ver) + if err != nil { + return err + } + ver.ID = vid + s.versions[vid] = ver + + ev := &ChangeEvent{ + DocID: docID, VersionID: vid, EventNo: 1, ToSnapshotID: id, + ChangeKind: ChangeImport, ActorKind: ActorTemplate, ActorRef: "template", + ActorSnapshot: actor, IdempotencyKey: fmt.Sprintf("template:%s:%s", docID, sha), + CreatedAt: time.Now().UTC(), + } + eid, err := s.insertEventTX(ctx, tx, ev) + if err != nil { + return err + } + ev.ID = eid + s.events[eid] = ev + s.idemp[ev.IdempotencyKey] = eid + + root := &DocRootPointers{ + DocID: docID, Kind: "controlled", ProjectID: projectID, WorkDir: workDir, + LedgerVersionID: &vid, HeadSnapshotID: &id, HeadRevisionNo: 0, + ContentSHA256: sha, LifecycleState: "drafting", RevisionSchemaVer: SchemaVersion, + } + s.roots[docID] = root + s.blobs[sha] = blob + return s.projectRootPointersTX(ctx, tx, root) + }) + if err != nil { + return nil, err + } + return s.mustStateLocked(docID), nil +} + +func (s *Service) persistHumanRevisionLocked(ctx context.Context, root *DocRootPointers, ver *Version, snap *Snapshot, ev *ChangeEvent, isBaseline bool) error { + if !s.usingPG() { + return nil + } + blob := s.blobs[snap.ContentSHA256] + return s.withTx(ctx, func(tx pgx.Tx) error { + if err := s.ensureBlobTX(ctx, tx, blob); err != nil { + return err + } + oldSnapID := snap.ID + id, err := s.insertSnapshotTX(ctx, tx, snap) + if err != nil { + return err + } + if oldSnapID != 0 && oldSnapID != id { + delete(s.snapshots, oldSnapID) + } + snap.ID = id + s.snapshots[id] = snap + + if isBaseline { + ver.BaselineSnapshotID = id + ver.HeadSnapshotID = id + oldVerID := ver.ID + vid, err := s.insertVersionTX(ctx, tx, ver) + if err != nil { + return err + } + if oldVerID != 0 && oldVerID != vid { + delete(s.versions, oldVerID) + } + ver.ID = vid + s.versions[vid] = ver + root.LedgerVersionID = &vid + root.HeadSnapshotID = &id + ev.VersionID = vid + ev.ToSnapshotID = id + } else { + ver.HeadSnapshotID = id + root.HeadSnapshotID = &id + if err := s.updateVersionTX(ctx, tx, ver); err != nil { + return err + } + ev.ToSnapshotID = id + ev.VersionID = ver.ID + } + + oldEv := ev.ID + eid, err := s.insertEventTX(ctx, tx, ev) + if err != nil { + return err + } + if oldEv != 0 && oldEv != eid { + delete(s.events, oldEv) + } + ev.ID = eid + s.events[eid] = ev + if ev.IdempotencyKey != "" { + s.idemp[ev.IdempotencyKey] = eid + } + return s.projectRootPointersTX(ctx, tx, root) + }) +} + +func (s *Service) persistAgentCandidateLocked(ctx context.Context, outSnap *Snapshot, rep *ChangeReport, c *AgentCandidate) error { + if !s.usingPG() { + return nil + } + blob := s.blobs[outSnap.ContentSHA256] + return s.withTx(ctx, func(tx pgx.Tx) error { + if err := s.ensureBlobTX(ctx, tx, blob); err != nil { + return err + } + oldSnap := outSnap.ID + sid, err := s.insertSnapshotTX(ctx, tx, outSnap) + if err != nil { + return err + } + if oldSnap != 0 && oldSnap != sid { + delete(s.snapshots, oldSnap) + } + outSnap.ID = sid + s.snapshots[sid] = outSnap + c.OutputSnapshotID = &sid + rep.ToSnapshotID = sid + + oldRep := rep.ID + rid, err := s.insertReportTX(ctx, tx, rep) + if err != nil { + return err + } + if oldRep != 0 && oldRep != rid { + delete(s.reports, oldRep) + } + rep.ID = rid + s.reports[rid] = rep + c.ChangeReportID = &rid + + oldC := c.ID + cid, err := s.insertCandidateTX(ctx, tx, c) + if err != nil { + return err + } + if oldC != 0 && oldC != cid { + delete(s.candidates, oldC) + } + c.ID = cid + s.candidates[cid] = c + return nil + }) +} + +func (s *Service) persistAgentCommitLocked(ctx context.Context, root *DocRootPointers, oldVer, newVer *Version, outSnap *Snapshot, ev *ChangeEvent, c *AgentCandidate) error { + if !s.usingPG() { + return nil + } + return s.withTx(ctx, func(tx pgx.Tx) error { + if err := s.updateVersionTX(ctx, tx, oldVer); err != nil { + return err + } + oldVerID := newVer.ID + vid, err := s.insertVersionTX(ctx, tx, newVer) + if err != nil { + return err + } + if oldVerID != 0 && oldVerID != vid { + delete(s.versions, oldVerID) + } + newVer.ID = vid + s.versions[vid] = newVer + root.LedgerVersionID = &vid + hid := outSnap.ID + root.HeadSnapshotID = &hid + + ev.VersionID = vid + ev.ToSnapshotID = outSnap.ID + oldEv := ev.ID + eid, err := s.insertEventTX(ctx, tx, ev) + if err != nil { + return err + } + if oldEv != 0 { + delete(s.events, oldEv) + } + ev.ID = eid + s.events[eid] = ev + + if err := s.updateCandidateTX(ctx, tx, c); err != nil { + return err + } + if newVer.ChangeReportID != nil { + _, _ = tx.Exec(ctx, `UPDATE document_versions SET change_report_id=$2 WHERE id=$1`, vid, *newVer.ChangeReportID) + } + return s.projectRootPointersTX(ctx, tx, root) + }) +} diff --git a/internal/docrevision/pg_persist_test.go b/internal/docrevision/pg_persist_test.go new file mode 100644 index 00000000..75ca261a --- /dev/null +++ b/internal/docrevision/pg_persist_test.go @@ -0,0 +1,62 @@ +package docrevision_test + +import ( + "context" + "testing" + + "chat-agent-system/internal/db" + "chat-agent-system/internal/docrevision" + + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestPGPersistSurvivesNewServiceInstance(t *testing.T) { + dsn := db.SkipIfNoTestDatabaseURL(t) + ctx := context.Background() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + + // Best-effort: skip if schema 00028 tables missing. + var n int + if err := pool.QueryRow(ctx, `SELECT COUNT(*) FROM information_schema.tables WHERE table_name='document_versions'`).Scan(&n); err != nil || n == 0 { + t.Skip("document_versions table missing; migrate 00028 required") + } + + docID := "doc_pg_persist_test" + _, _ = pool.Exec(ctx, `DELETE FROM document_change_events WHERE doc_id=$1`, docID) + _, _ = pool.Exec(ctx, `DELETE FROM document_agent_candidates WHERE doc_id=$1`, docID) + _, _ = pool.Exec(ctx, `DELETE FROM document_change_reports WHERE doc_id=$1`, docID) + _, _ = pool.Exec(ctx, `DELETE FROM document_versions WHERE doc_id=$1`, docID) + _, _ = pool.Exec(ctx, `DELETE FROM document_snapshots WHERE doc_id=$1`, docID) + + svc1 := docrevision.NewService(pool) + st, err := svc1.OpenTemplateVersion(ctx, docID, "# persist\n\nhello\n", docrevision.ActorSnapshot{DisplayName: "t"}, 1, "ws/persist") + if err != nil { + t.Fatal(err) + } + _, err = svc1.CommitHumanRevision(ctx, docrevision.HumanRevisionRequest{ + DocID: docID, Markdown: "# persist\n\nhello edited\n", + ExpectedHeadSnapshot: &st.HeadSnapshotID, ExpectedRevisionNo: &st.HeadRevisionNo, + IdempotencyKey: "pg-e1", ActorUserID: 1, LifecycleState: "drafting", WorkDir: "ws/persist", ProjectID: 1, + }) + if err != nil { + t.Fatal(err) + } + + // Simulate process restart: brand-new service, same pool. + svc2 := docrevision.NewService(pool) + got, err := svc2.GetRevisionState(ctx, docID) + if err != nil { + t.Fatalf("hydrate after restart: %v", err) + } + if got.HeadRevisionNo != 1 { + t.Fatalf("want revision 1 after restart, got %+v", got) + } + ov, err := svc2.ManualOverlay(ctx, docID) + if err != nil || ov == nil { + t.Fatalf("overlay after restart: %v", err) + } +} diff --git a/internal/docrevision/pg_store.go b/internal/docrevision/pg_store.go new file mode 100644 index 00000000..b0157f97 --- /dev/null +++ b/internal/docrevision/pg_store.go @@ -0,0 +1,471 @@ +package docrevision + +import ( + "context" + "encoding/json" + "errors" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// ErrNeedsContent indicates the document has no server-side body to bootstrap a ledger from. +var ErrNeedsContent = errors.New("docrevision: needs content to bootstrap") + +const ErrCodeNeedsContent = "DOCUMENT_REVISION_NEEDS_CONTENT" + +// BootstrapRequest supplies the fact-source markdown used for first-access ledger creation. +type BootstrapRequest struct { + DocID string + Kind string // controlled|freeform + Markdown string + ProjectID int64 + WorkDir string + LifecycleState string + Actor ActorSnapshot + Origin string // template|migration +} + +func (s *Service) usingPG() bool { + return s != nil && s.pool != nil +} + +// hydrateDocLocked loads ledger rows for docID from PostgreSQL into memory maps. +// Caller must hold s.mu. +func (s *Service) hydrateDocLocked(ctx context.Context, docID string) error { + if !s.usingPG() || docID == "" { + return nil + } + if root, ok := s.roots[docID]; ok && root.LedgerVersionID != nil { + return nil + } + + rows, err := s.pool.Query(ctx, ` + SELECT id, doc_id, content_sha256, parent_snapshot_id, block_map, source_kind, + actor_user_id, actor_ref, actor_snapshot, source_run_id, created_at + FROM document_snapshots WHERE doc_id = $1 ORDER BY id`, docID) + if err != nil { + return err + } + defer rows.Close() + + type snapRow struct { + snap Snapshot + } + var snaps []snapRow + shas := map[string]struct{}{} + for rows.Next() { + var sn Snapshot + var blockRaw []byte + var actorRaw []byte + if err := rows.Scan(&sn.ID, &sn.DocID, &sn.ContentSHA256, &sn.ParentSnapshotID, &blockRaw, + &sn.SourceKind, &sn.ActorUserID, &sn.ActorRef, &actorRaw, &sn.SourceRunID, &sn.CreatedAt); err != nil { + return err + } + _ = json.Unmarshal(blockRaw, &sn.BlockMap) + _ = json.Unmarshal(actorRaw, &sn.ActorSnapshot) + snaps = append(snaps, snapRow{snap: sn}) + shas[sn.ContentSHA256] = struct{}{} + } + if err := rows.Err(); err != nil { + return err + } + if len(snaps) == 0 { + return nil + } + + for sha := range shas { + var md string + var byteSize int64 + var created time.Time + err := s.pool.QueryRow(ctx, ` + SELECT content_markdown, byte_size, created_at FROM document_content_blobs WHERE content_sha256 = $1`, sha). + Scan(&md, &byteSize, &created) + if err != nil { + return err + } + s.blobs[sha] = &ContentBlob{ContentSHA256: sha, ContentMarkdown: md, ByteSize: byteSize, CreatedAt: created} + } + for i := range snaps { + cp := snaps[i].snap + s.snapshots[cp.ID] = &cp + if cp.ID > s.nextSnapID { + s.nextSnapID = cp.ID + } + } + + vrows, err := s.pool.Query(ctx, ` + SELECT id, doc_id, version_no, status, origin_kind, baseline_snapshot_id, head_snapshot_id, + human_revision_count, source_run_id, change_report_id, opened_at, sealed_at + FROM document_versions WHERE doc_id = $1 ORDER BY version_no`, docID) + if err != nil { + return err + } + defer vrows.Close() + var openVer *Version + for vrows.Next() { + var v Version + if err := vrows.Scan(&v.ID, &v.DocID, &v.VersionNo, &v.Status, &v.OriginKind, &v.BaselineSnapshotID, + &v.HeadSnapshotID, &v.HumanRevisionCount, &v.SourceRunID, &v.ChangeReportID, &v.OpenedAt, &v.SealedAt); err != nil { + return err + } + cp := v + s.versions[cp.ID] = &cp + if cp.ID > s.nextVerID { + s.nextVerID = cp.ID + } + if cp.Status == VersionOpen { + openVer = &cp + } + } + if err := vrows.Err(); err != nil { + return err + } + + erows, err := s.pool.Query(ctx, ` + SELECT id, doc_id, version_id, event_no, from_snapshot_id, to_snapshot_id, change_kind, actor_kind, + actor_user_id, actor_ref, actor_snapshot, change_set, edit_session_id, request_id, idempotency_key, created_at + FROM document_change_events WHERE doc_id = $1 ORDER BY event_no`, docID) + if err != nil { + return err + } + defer erows.Close() + for erows.Next() { + var ev ChangeEvent + var actorRaw, changeRaw []byte + if err := erows.Scan(&ev.ID, &ev.DocID, &ev.VersionID, &ev.EventNo, &ev.FromSnapshotID, &ev.ToSnapshotID, + &ev.ChangeKind, &ev.ActorKind, &ev.ActorUserID, &ev.ActorRef, &actorRaw, &changeRaw, + &ev.EditSessionID, &ev.RequestID, &ev.IdempotencyKey, &ev.CreatedAt); err != nil { + return err + } + _ = json.Unmarshal(actorRaw, &ev.ActorSnapshot) + _ = json.Unmarshal(changeRaw, &ev.ChangeSet) + cp := ev + s.events[cp.ID] = &cp + if cp.IdempotencyKey != "" { + s.idemp[cp.IdempotencyKey] = cp.ID + } + if cp.ID > s.nextEvtID { + s.nextEvtID = cp.ID + } + } + if err := erows.Err(); err != nil { + return err + } + + rrows, err := s.pool.Query(ctx, ` + SELECT id, doc_id, from_snapshot_id, to_snapshot_id, schema_version, summary, risk_level, risk_items, + project_impacts, document_impacts, open_questions, analysis_status, model_snapshot, prompt_version, + deterministic_facts, corrects_report_id, raw_json, created_at + FROM document_change_reports WHERE doc_id = $1`, docID) + if err != nil { + return err + } + defer rrows.Close() + for rrows.Next() { + var rep ChangeReport + var sumRaw, riskRaw, projRaw, docRaw, qRaw, factRaw []byte + if err := rrows.Scan(&rep.ID, &rep.DocID, &rep.FromSnapshotID, &rep.ToSnapshotID, &rep.SchemaVersion, + &sumRaw, &rep.RiskLevel, &riskRaw, &projRaw, &docRaw, &qRaw, &rep.AnalysisStatus, + &rep.ModelSnapshot, &rep.PromptVersion, &factRaw, &rep.CorrectsReportID, &rep.RawJSON, &rep.CreatedAt); err != nil { + return err + } + _ = json.Unmarshal(sumRaw, &rep.Summary) + _ = json.Unmarshal(riskRaw, &rep.RiskItems) + _ = json.Unmarshal(projRaw, &rep.ProjectImpacts) + _ = json.Unmarshal(docRaw, &rep.DocumentImpacts) + _ = json.Unmarshal(qRaw, &rep.OpenQuestions) + _ = json.Unmarshal(factRaw, &rep.DeterministicFacts) + cp := rep + s.reports[cp.ID] = &cp + if cp.ID > s.nextRepID { + s.nextRepID = cp.ID + } + } + if err := rrows.Err(); err != nil { + return err + } + + crows, err := s.pool.Query(ctx, ` + SELECT id, doc_id, input_snapshot_id, output_snapshot_id, run_id, proposal_id, status, authorized_scope, + quality_result, change_report_id, actor_kind, actor_snapshot, created_at, decided_at + FROM document_agent_candidates WHERE doc_id = $1`, docID) + if err != nil { + return err + } + defer crows.Close() + for crows.Next() { + var c AgentCandidate + var actorRaw []byte + if err := crows.Scan(&c.ID, &c.DocID, &c.InputSnapshotID, &c.OutputSnapshotID, &c.RunID, &c.ProposalID, + &c.Status, &c.AuthorizedScope, &c.QualityResult, &c.ChangeReportID, &c.ActorKind, &actorRaw, + &c.CreatedAt, &c.DecidedAt); err != nil { + return err + } + _ = json.Unmarshal(actorRaw, &c.ActorSnapshot) + cp := c + s.candidates[cp.ID] = &cp + if cp.ID > s.nextCandID { + s.nextCandID = cp.ID + } + } + if err := crows.Err(); err != nil { + return err + } + + root := &DocRootPointers{DocID: docID, RevisionSchemaVer: SchemaVersion, Kind: "controlled", LifecycleState: "drafting"} + if openVer != nil { + vid := openVer.ID + hid := openVer.HeadSnapshotID + root.LedgerVersionID = &vid + root.HeadSnapshotID = &hid + root.HeadRevisionNo = openVer.HumanRevisionCount + if head := s.snapshots[openVer.HeadSnapshotID]; head != nil { + root.ContentSHA256 = head.ContentSHA256 + } + } + + var ledgerID, headID *int64 + var headRev int + var workDir, lifecycle string + var projectID int64 + err = s.pool.QueryRow(ctx, ` + SELECT ledger_version_id, head_snapshot_id, head_revision_no, COALESCE(work_dir,''), COALESCE(current_state,'drafting'), project_id + FROM document_records WHERE doc_id = $1`, docID). + Scan(&ledgerID, &headID, &headRev, &workDir, &lifecycle, &projectID) + if err == nil { + root.Kind = "controlled" + root.WorkDir = workDir + root.LifecycleState = lifecycle + root.ProjectID = projectID + if ledgerID != nil { + root.LedgerVersionID = ledgerID + } + if headID != nil { + root.HeadSnapshotID = headID + } + root.HeadRevisionNo = headRev + } else if errors.Is(err, pgx.ErrNoRows) { + var fLedger, fHead *int64 + var fRev int + var fProj int64 + ferr := s.pool.QueryRow(ctx, ` + SELECT ledger_version_id, head_snapshot_id, head_revision_no, project_id + FROM freeform_documents WHERE doc_id = $1 AND NOT is_deleted`, docID). + Scan(&fLedger, &fHead, &fRev, &fProj) + if ferr == nil { + root.Kind = "freeform" + root.LifecycleState = "drafting" + root.ProjectID = fProj + if fLedger != nil { + root.LedgerVersionID = fLedger + } + if fHead != nil { + root.HeadSnapshotID = fHead + } + root.HeadRevisionNo = fRev + } else if !errors.Is(ferr, pgx.ErrNoRows) { + return ferr + } + } else { + return err + } + + if root.LedgerVersionID != nil { + s.roots[docID] = root + } + return nil +} + +func (s *Service) ensureBlobTX(ctx context.Context, tx pgx.Tx, blob *ContentBlob) error { + _, err := tx.Exec(ctx, ` + INSERT INTO document_content_blobs (content_sha256, content_markdown, byte_size, created_at) + VALUES ($1, $2, $3, $4) + ON CONFLICT (content_sha256) DO NOTHING`, + blob.ContentSHA256, blob.ContentMarkdown, blob.ByteSize, blob.CreatedAt) + return err +} + +func (s *Service) insertSnapshotTX(ctx context.Context, tx pgx.Tx, sn *Snapshot) (int64, error) { + blockRaw, _ := json.Marshal(sn.BlockMap) + if blockRaw == nil { + blockRaw = []byte("[]") + } + actorRaw, _ := json.Marshal(sn.ActorSnapshot) + if actorRaw == nil { + actorRaw = []byte("{}") + } + var id int64 + err := tx.QueryRow(ctx, ` + INSERT INTO document_snapshots ( + doc_id, content_sha256, parent_snapshot_id, block_map, source_kind, + actor_user_id, actor_ref, actor_snapshot, source_run_id, created_at + ) VALUES ($1,$2,$3,$4::jsonb,$5,$6,$7,$8::jsonb,$9,$10) RETURNING id`, + sn.DocID, sn.ContentSHA256, sn.ParentSnapshotID, blockRaw, sn.SourceKind, + sn.ActorUserID, sn.ActorRef, actorRaw, sn.SourceRunID, sn.CreatedAt, + ).Scan(&id) + return id, err +} + +func (s *Service) insertVersionTX(ctx context.Context, tx pgx.Tx, v *Version) (int64, error) { + var id int64 + err := tx.QueryRow(ctx, ` + INSERT INTO document_versions ( + doc_id, version_no, status, origin_kind, baseline_snapshot_id, head_snapshot_id, + human_revision_count, source_run_id, change_report_id, opened_at, sealed_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) RETURNING id`, + v.DocID, v.VersionNo, v.Status, v.OriginKind, v.BaselineSnapshotID, v.HeadSnapshotID, + v.HumanRevisionCount, v.SourceRunID, v.ChangeReportID, v.OpenedAt, v.SealedAt, + ).Scan(&id) + return id, err +} + +func (s *Service) updateVersionTX(ctx context.Context, tx pgx.Tx, v *Version) error { + _, err := tx.Exec(ctx, ` + UPDATE document_versions SET status=$2, head_snapshot_id=$3, human_revision_count=$4, + change_report_id=$5, sealed_at=$6 WHERE id=$1`, + v.ID, v.Status, v.HeadSnapshotID, v.HumanRevisionCount, v.ChangeReportID, v.SealedAt) + return err +} + +func (s *Service) insertEventTX(ctx context.Context, tx pgx.Tx, ev *ChangeEvent) (int64, error) { + actorRaw, _ := json.Marshal(ev.ActorSnapshot) + if actorRaw == nil { + actorRaw = []byte("{}") + } + changeRaw, _ := json.Marshal(ev.ChangeSet) + if changeRaw == nil { + changeRaw = []byte("[]") + } + var id int64 + err := tx.QueryRow(ctx, ` + INSERT INTO document_change_events ( + doc_id, version_id, event_no, from_snapshot_id, to_snapshot_id, change_kind, actor_kind, + actor_user_id, actor_ref, actor_snapshot, change_set, edit_session_id, request_id, idempotency_key, created_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10::jsonb,$11::jsonb,$12,$13,$14,$15) RETURNING id`, + ev.DocID, ev.VersionID, ev.EventNo, ev.FromSnapshotID, ev.ToSnapshotID, ev.ChangeKind, ev.ActorKind, + ev.ActorUserID, ev.ActorRef, actorRaw, changeRaw, ev.EditSessionID, ev.RequestID, ev.IdempotencyKey, ev.CreatedAt, + ).Scan(&id) + return id, err +} + +func (s *Service) insertReportTX(ctx context.Context, tx pgx.Tx, rep *ChangeReport) (int64, error) { + sumRaw, _ := json.Marshal(rep.Summary) + if sumRaw == nil { + sumRaw = []byte("[]") + } + riskRaw, _ := json.Marshal(rep.RiskItems) + if riskRaw == nil { + riskRaw = []byte("[]") + } + projRaw, _ := json.Marshal(rep.ProjectImpacts) + if projRaw == nil { + projRaw = []byte("{}") + } + docRaw, _ := json.Marshal(rep.DocumentImpacts) + if docRaw == nil { + docRaw = []byte("[]") + } + qRaw, _ := json.Marshal(rep.OpenQuestions) + if qRaw == nil { + qRaw = []byte("[]") + } + factRaw, _ := json.Marshal(rep.DeterministicFacts) + if factRaw == nil { + factRaw = []byte("{}") + } + var id int64 + err := tx.QueryRow(ctx, ` + INSERT INTO document_change_reports ( + doc_id, from_snapshot_id, to_snapshot_id, schema_version, summary, risk_level, risk_items, + project_impacts, document_impacts, open_questions, analysis_status, model_snapshot, prompt_version, + deterministic_facts, corrects_report_id, raw_json, created_at + ) VALUES ($1,$2,$3,$4,$5::jsonb,$6,$7::jsonb,$8::jsonb,$9::jsonb,$10::jsonb,$11,$12,$13,$14::jsonb,$15,$16,$17) + RETURNING id`, + rep.DocID, rep.FromSnapshotID, rep.ToSnapshotID, rep.SchemaVersion, sumRaw, rep.RiskLevel, riskRaw, + projRaw, docRaw, qRaw, rep.AnalysisStatus, rep.ModelSnapshot, rep.PromptVersion, + factRaw, rep.CorrectsReportID, rep.RawJSON, rep.CreatedAt, + ).Scan(&id) + return id, err +} + +func (s *Service) insertCandidateTX(ctx context.Context, tx pgx.Tx, c *AgentCandidate) (int64, error) { + actorRaw, _ := json.Marshal(c.ActorSnapshot) + if actorRaw == nil { + actorRaw = []byte("{}") + } + var id int64 + err := tx.QueryRow(ctx, ` + INSERT INTO document_agent_candidates ( + doc_id, input_snapshot_id, output_snapshot_id, run_id, proposal_id, status, authorized_scope, + quality_result, change_report_id, actor_kind, actor_snapshot, created_at, decided_at + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::jsonb,$12,$13) RETURNING id`, + c.DocID, c.InputSnapshotID, c.OutputSnapshotID, c.RunID, c.ProposalID, c.Status, c.AuthorizedScope, + c.QualityResult, c.ChangeReportID, c.ActorKind, actorRaw, c.CreatedAt, c.DecidedAt, + ).Scan(&id) + return id, err +} + +func (s *Service) updateCandidateTX(ctx context.Context, tx pgx.Tx, c *AgentCandidate) error { + _, err := tx.Exec(ctx, ` + UPDATE document_agent_candidates SET status=$2, decided_at=$3, change_report_id=$4, output_snapshot_id=$5 + WHERE id=$1`, c.ID, c.Status, c.DecidedAt, c.ChangeReportID, c.OutputSnapshotID) + return err +} + +func (s *Service) projectRootPointersTX(ctx context.Context, tx pgx.Tx, root *DocRootPointers) error { + if root == nil { + return nil + } + switch root.Kind { + case "freeform": + _, err := tx.Exec(ctx, ` + UPDATE freeform_documents + SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, + revision_schema_version=$5, row_revision=COALESCE(row_revision,0)+1 + WHERE doc_id=$1`, + root.DocID, root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer) + return err + default: + _, err := tx.Exec(ctx, ` + UPDATE document_records + SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4, + revision_schema_version=$5, row_revision=COALESCE(row_revision,0)+1, + content_sha256=COALESCE(NULLIF($6,''), content_sha256) + WHERE doc_id=$1`, + root.DocID, root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, + root.RevisionSchemaVer, root.ContentSHA256) + return err + } +} + +func (s *Service) withTx(ctx context.Context, fn func(pgx.Tx) error) error { + if !s.usingPG() { + return nil + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(ctx) + if err := fn(tx); err != nil { + return err + } + return tx.Commit(ctx) +} + +// Pool exposes the underlying pool for integration tests. +func (s *Service) Pool() *pgxpool.Pool { return s.pool } + +// remapSnapshotID updates parent pointers after a snapshot receives its real PG id. +func (s *Service) remapSnapshotID(oldID, newID int64) { + if oldID == 0 || oldID == newID { + return + } + if sn := s.snapshots[oldID]; sn != nil { + delete(s.snapshots, oldID) + sn.ID = newID + s.snapshots[newID] = sn + } +} diff --git a/internal/docrevision/risk.go b/internal/docrevision/risk.go index e2055822..2263b2d5 100644 --- a/internal/docrevision/risk.go +++ b/internal/docrevision/risk.go @@ -110,6 +110,66 @@ func ValidateChangeReportJSON(raw string, fromSHA, toSHA string, ruleRisk string return rep, nil } +// BuildDeterministicChangeReportJSON builds a non-fabricated report from diff + rule risk + lineage. +// Narrative PI text is intentionally absent; analysis_status is fallback. +func BuildDeterministicChangeReportJSON(fromSHA, toSHA, ruleRisk string, impacts []DocImpact, diff DiffResult) string { + summaries := make([]string, 0, 4) + summaries = append(summaries, fmt.Sprintf("确定性差异:+%d / -%d / ~%d 处块级变更", + diff.Summary.Inserts, diff.Summary.Deletes, diff.Summary.Modifies)) + for i, h := range diff.Hunks { + if i >= 3 { + break + } + sec := h.Section + if sec == "" { + sec = h.Kind + } + summaries = append(summaries, fmt.Sprintf("%s:%s", h.Kind, sec)) + } + if fromSHA == toSHA { + summaries = []string{"内容哈希未变化(agent noop 候选)"} + } + riskItems := []RiskItem{} + if ruleRisk == RiskHigh || ruleRisk == RiskCritical { + riskItems = append(riskItems, RiskItem{ + Category: "rule_floor", + Description: "规则引擎根据敏感删除/数值/下游影响抬升风险", + Mitigation: "由编制负责人确认后提交", + }) + } + proj := map[string]any{ + "summary": fmt.Sprintf("下游受影响文档 %d 份", len(impacts)), + } + payload := map[string]any{ + "schema_version": 1, + "summary": summaries, + "risk": map[string]any{"level": ruleRisk, "items": riskItems}, + "project_impact": proj, + "document_impacts": impacts, + "open_questions": []string{ + "业务叙事摘要待真实 PI 变更报告链路补齐(DOCREV-RT-PI-NARRATIVE)", + }, + } + raw, _ := json.Marshal(payload) + // Mark fallback via validator path: empty open_questions would be complete; + // we keep open_questions and let caller set AnalysisFallback after validate when desired. + _ = raw + // Force fallback semantics by returning JSON that validates, then caller may override status. + rep, err := ValidateChangeReportJSON(string(raw), fromSHA, toSHA, ruleRisk, impacts, diff.Hunks) + if err != nil || rep == nil { + fb := fallbackReport(fromSHA, toSHA, ruleRisk, impacts, diff.Hunks, "deterministic build") + fb.AnalysisStatus = AnalysisFallback + out, _ := json.Marshal(map[string]any{ + "schema_version": 1, "summary": fb.Summary, + "risk": map[string]any{"level": fb.RiskLevel, "items": fb.RiskItems}, + "project_impact": proj, "document_impacts": impacts, "open_questions": fb.OpenQuestions, + }) + return string(out) + } + // Re-encode with explicit fallback marker in open_questions already present. + return string(raw) +} + func RiskUnassessedAsHigh() string { return RiskHigh } func fallbackReport(fromSHA, toSHA, ruleRisk string, impacts []DocImpact, hunks []DiffHunk, reason string) *ChangeReport { diff --git a/internal/docrevision/service.go b/internal/docrevision/service.go index b837b4c5..107a7102 100644 --- a/internal/docrevision/service.go +++ b/internal/docrevision/service.go @@ -2,12 +2,15 @@ package docrevision import ( "context" + "errors" "fmt" "os" "path/filepath" + "strings" "sync" "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -34,6 +37,12 @@ type Service struct { // Outbox hooks (Wave 5); optional. OnOutbox func(ctx context.Context, eventType string, payload map[string]any) error + // ImpactResolver returns deterministic downstream impacts (lineage); optional. + ImpactResolver func(ctx context.Context, docID string) []DocImpact + + // DataDir is the environment data root used to resolve relative WorkDir paths. + DataDir string + // Feature flags LedgerWrite bool LedgerRead bool @@ -79,9 +88,11 @@ func (s *Service) EnsureRoot(docID, kind string, projectID int64, lifecycle stri } func (s *Service) GetRevisionState(ctx context.Context, docID string) (*RevisionState, error) { - _ = ctx s.mu.Lock() defer s.mu.Unlock() + if err := s.hydrateDocLocked(ctx, docID); err != nil { + return nil, err + } root, ok := s.roots[docID] if !ok || root.LedgerVersionID == nil { return nil, ErrNotFound @@ -109,8 +120,39 @@ func (s *Service) GetRevisionState(ctx context.Context, docID string) (*Revision }, nil } +// EnsureBootstrap creates a ledger from fact-source markdown when none exists (idempotent). +func (s *Service) EnsureBootstrap(ctx context.Context, req BootstrapRequest) (*RevisionState, error) { + if st, err := s.GetRevisionState(ctx, req.DocID); err == nil { + return st, nil + } else if err != nil && !errors.Is(err, ErrNotFound) { + return nil, err + } + md := strings.TrimSpace(req.Markdown) + if md == "" { + return nil, ErrNeedsContent + } + kind := req.Kind + if kind == "" { + kind = "controlled" + } + life := req.LifecycleState + if life == "" { + life = "drafting" + } + st, err := s.OpenTemplateVersion(ctx, req.DocID, md, req.Actor, req.ProjectID, req.WorkDir) + if err != nil { + return nil, err + } + s.mu.Lock() + if root := s.roots[req.DocID]; root != nil { + root.Kind = kind + root.LifecycleState = life + } + s.mu.Unlock() + return st, nil +} + func (s *Service) CommitHumanRevision(ctx context.Context, req HumanRevisionRequest) (*HumanRevisionResult, error) { - _ = ctx if req.IdempotencyKey == "" { return nil, fmt.Errorf("docrevision: idempotency key required") } @@ -125,6 +167,9 @@ func (s *Service) CommitHumanRevision(ctx context.Context, req HumanRevisionRequ s.mu.Lock() defer s.mu.Unlock() + if err := s.hydrateDocLocked(ctx, req.DocID); err != nil { + return nil, err + } if eid, ok := s.idemp[req.IdempotencyKey]; ok { ev := s.events[eid] @@ -135,9 +180,19 @@ func (s *Service) CommitHumanRevision(ctx context.Context, req HumanRevisionRequ root := s.roots[req.DocID] if root == nil { - root = &DocRootPointers{DocID: req.DocID, Kind: "controlled", LifecycleState: req.LifecycleState, ProjectID: req.ProjectID, RevisionSchemaVer: SchemaVersion, WorkDir: req.WorkDir} + kind := "controlled" + if req.WorkDir == "" && req.ProjectID > 0 { + kind = "controlled" + } + root = &DocRootPointers{DocID: req.DocID, Kind: kind, LifecycleState: req.LifecycleState, ProjectID: req.ProjectID, RevisionSchemaVer: SchemaVersion, WorkDir: req.WorkDir} s.roots[req.DocID] = root } + if req.WorkDir != "" { + root.WorkDir = req.WorkDir + } + if req.ProjectID > 0 { + root.ProjectID = req.ProjectID + } if root.LedgerVersionID == nil { if !req.AllowCreateBaseline { @@ -151,6 +206,9 @@ func (s *Service) CommitHumanRevision(ctx context.Context, req HumanRevisionRequ root.ContentSHA256 = sha root.RowRevision++ ev := s.putEventLocked(req.DocID, ver.ID, 1, nil, snap.ID, ChangeImport, ActorHuman, &req.ActorUserID, "", req.ActorSnapshot, nil, req.EditSessionID, req.RequestID, req.IdempotencyKey) + if err := s.persistHumanRevisionLocked(ctx, root, ver, snap, ev, true); err != nil { + return nil, err + } st := s.mustStateLocked(req.DocID) return &HumanRevisionResult{State: *st, Event: *ev}, nil } @@ -177,7 +235,6 @@ func (s *Service) CommitHumanRevision(ctx context.Context, req HumanRevisionRequ parentID := ver.HeadSnapshotID parent := s.snapshots[parentID] if parent.ContentSHA256 == sha { - // no-op: return current without new snapshot/version st := s.mustStateLocked(req.DocID) overlay := s.manualOverlayLocked(req.DocID) return &HumanRevisionResult{State: *st, Overlay: overlay}, nil @@ -199,24 +256,31 @@ func (s *Service) CommitHumanRevision(ctx context.Context, req HumanRevisionRequ root.RowRevision++ eventNo := s.nextEventNoLocked(req.DocID) ev := s.putEventLocked(req.DocID, ver.ID, eventNo, &parentID, snap.ID, ChangeHumanRevision, ActorHuman, &req.ActorUserID, "", req.ActorSnapshot, diff.Hunks, req.EditSessionID, req.RequestID, req.IdempotencyKey) + if err := s.persistHumanRevisionLocked(ctx, root, ver, snap, ev, false); err != nil { + return nil, err + } st := s.mustStateLocked(req.DocID) - overlay := DiffResult{FromSnapshotID: ver.BaselineSnapshotID, ToSnapshotID: snap.ID, Summary: diff.Summary, Hunks: diff.Hunks} baseMD := s.blobs[s.snapshots[ver.BaselineSnapshotID].ContentSHA256].ContentMarkdown - overlay = ComputeDiff(baseMD, canon, s.snapshots[ver.BaselineSnapshotID].BlockMap, snap.BlockMap) + overlay := ComputeDiff(baseMD, canon, s.snapshots[ver.BaselineSnapshotID].BlockMap, snap.BlockMap) overlay.FromSnapshotID = ver.BaselineSnapshotID overlay.ToSnapshotID = snap.ID return &HumanRevisionResult{State: *st, Event: *ev, Overlay: overlay}, nil } func (s *Service) OpenTemplateVersion(ctx context.Context, docID string, markdown string, actor ActorSnapshot, projectID int64, workDir string) (*RevisionState, error) { - _ = ctx canon := CanonicalizeMarkdown(markdown) sha := SHA256Hex(canon) s.mu.Lock() defer s.mu.Unlock() + if err := s.hydrateDocLocked(ctx, docID); err != nil { + return nil, err + } if root, ok := s.roots[docID]; ok && root.LedgerVersionID != nil { return s.mustStateLocked(docID), nil } + if s.usingPG() { + return s.openTemplatePGLocked(ctx, docID, canon, sha, actor, projectID, workDir) + } snap := s.putSnapshotLocked(docID, canon, sha, nil, SourceTemplate, nil, "template", actor, "", nil) ver := s.openVersionLocked(docID, OriginTemplate, snap.ID, snap.ID, "") root := &DocRootPointers{ @@ -230,7 +294,6 @@ func (s *Service) OpenTemplateVersion(ctx context.Context, docID string, markdow } func (s *Service) CreateAgentCandidate(ctx context.Context, req AgentCandidateRequest) (*AgentCandidate, *ChangeReport, error) { - _ = ctx if !s.AgentCommit { return nil, nil, fmt.Errorf("docrevision: agent commit disabled") } @@ -242,6 +305,9 @@ func (s *Service) CreateAgentCandidate(ctx context.Context, req AgentCandidateRe s.mu.Lock() defer s.mu.Unlock() + if err := s.hydrateDocLocked(ctx, req.DocID); err != nil { + return nil, nil, err + } root := s.roots[req.DocID] if root == nil || root.LedgerVersionID == nil || root.HeadSnapshotID == nil { @@ -257,23 +323,27 @@ func (s *Service) CreateAgentCandidate(ctx context.Context, req AgentCandidateRe inMD := s.blobs[inSnap.ContentSHA256].ContentMarkdown outSnap := s.putSnapshotLocked(req.DocID, canon, outSHA, &req.InputSnapshotID, req.ActorKind, req.OnBehalfOf, req.RunID, req.ActorSnapshot, req.RunID, nil) diff := ComputeDiff(inMD, canon, inSnap.BlockMap, outSnap.BlockMap) - impacts := s.impactLocked(req.DocID) + impacts := s.impactLocked(ctx, req.DocID) ruleRisk := RuleRiskFloor(diff.Hunks, impacts) - rep, err := ValidateChangeReportJSON(req.ReportJSON, inSnap.ContentSHA256, outSHA, ruleRisk, impacts, diff.Hunks) + reportJSON := req.ReportJSON + if strings.TrimSpace(reportJSON) == "" { + reportJSON = BuildDeterministicChangeReportJSON(inSnap.ContentSHA256, outSHA, ruleRisk, impacts, diff) + } + rep, err := ValidateChangeReportJSON(reportJSON, inSnap.ContentSHA256, outSHA, ruleRisk, impacts, diff.Hunks) status := CandidateReady if err != nil { status = CandidateReady rep.AnalysisStatus = AnalysisUnassessed } + if strings.TrimSpace(req.ReportJSON) == "" && rep != nil && rep.AnalysisStatus == AnalysisComplete { + rep.AnalysisStatus = AnalysisFallback + } if NeedsConfirmation(rep.RiskLevel) { status = CandidateNeedsConfirmation } if *root.HeadSnapshotID != req.InputSnapshotID { status = CandidateStale } - if inSnap.ContentSHA256 == outSHA { - // noop path handled at commit - } s.nextRepID++ rep.ID = s.nextRepID rep.DocID = req.DocID @@ -290,6 +360,9 @@ func (s *Service) CreateAgentCandidate(ctx context.Context, req AgentCandidateRe ActorKind: req.ActorKind, ActorSnapshot: req.ActorSnapshot, CreatedAt: time.Now().UTC(), } s.candidates[c.ID] = c + if err := s.persistAgentCandidateLocked(ctx, outSnap, rep, c); err != nil { + return nil, nil, err + } if s.Notify && s.OnOutbox != nil && status == CandidateNeedsConfirmation { _ = s.OnOutbox(ctx, "document.agent_candidate_ready", map[string]any{ "doc_id": req.DocID, "candidate_id": c.ID, "risk": rep.RiskLevel, @@ -302,10 +375,18 @@ func (s *Service) CreateAgentCandidate(ctx context.Context, req AgentCandidateRe } func (s *Service) CommitAgentVersion(ctx context.Context, candidateID int64, dec CommitAgentDecision) (*RevisionState, error) { - _ = ctx s.mu.Lock() defer s.mu.Unlock() c, ok := s.candidates[candidateID] + if !ok { + // try hydrate by scanning — candidate may only be in PG; load via unknown doc is hard. + // Caller should hydrate doc first; fallback: search all not possible. Return not found. + return nil, ErrNotFound + } + if err := s.hydrateDocLocked(ctx, c.DocID); err != nil { + return nil, err + } + c, ok = s.candidates[candidateID] if !ok { return nil, ErrNotFound } @@ -342,19 +423,29 @@ func (s *Service) CommitAgentVersion(ctx context.Context, candidateID int64, dec now := time.Now().UTC() c.Status = CandidateRejected c.DecidedAt = &now + if s.usingPG() { + _ = s.withTx(ctx, func(tx pgx.Tx) error { return s.updateCandidateTX(ctx, tx, c) }) + } return s.mustStateLocked(c.DocID), nil } if NeedsConfirmation(rep.RiskLevel) && !dec.Confirmed { return nil, ErrNeedsConfirmation } if inSnap.ContentSHA256 == outSnap.ContentSHA256 { - // agent noop: event only ver := s.versions[*root.LedgerVersionID] eventNo := s.nextEventNoLocked(c.DocID) - _ = s.putEventLocked(c.DocID, ver.ID, eventNo, &c.InputSnapshotID, outSnap.ID, ChangeAgentNoop, c.ActorKind, dec.ActorUserIDPtr(), c.RunID, c.ActorSnapshot, nil, "", "", fmt.Sprintf("agent_noop:%s:%s", c.RunID, outSnap.ContentSHA256)) + ev := s.putEventLocked(c.DocID, ver.ID, eventNo, &c.InputSnapshotID, outSnap.ID, ChangeAgentNoop, c.ActorKind, dec.ActorUserIDPtr(), c.RunID, c.ActorSnapshot, nil, "", "", fmt.Sprintf("agent_noop:%s:%s", c.RunID, outSnap.ContentSHA256)) now := time.Now().UTC() c.Status = CandidateCommitted c.DecidedAt = &now + if s.usingPG() { + _ = s.withTx(ctx, func(tx pgx.Tx) error { + if _, err := s.insertEventTX(ctx, tx, ev); err != nil { + return err + } + return s.updateCandidateTX(ctx, tx, c) + }) + } return s.mustStateLocked(c.DocID), nil } @@ -362,7 +453,7 @@ func (s *Service) CommitAgentVersion(ctx context.Context, candidateID int64, dec now := time.Now().UTC() oldVer.Status = VersionSealed oldVer.SealedAt = &now - oldVer.HeadSnapshotID = c.InputSnapshotID // seal at input (= previous head) + oldVer.HeadSnapshotID = c.InputSnapshotID origin := OriginAgent if c.ActorKind == ActorAgentSquad { @@ -380,10 +471,14 @@ func (s *Service) CommitAgentVersion(ctx context.Context, candidateID int64, dec outMD := s.blobs[outSnap.ContentSHA256].ContentMarkdown diff := ComputeDiff(inMD, outMD, inSnap.BlockMap, outSnap.BlockMap) eventNo := s.nextEventNoLocked(c.DocID) - _ = s.putEventLocked(c.DocID, newVer.ID, eventNo, &c.InputSnapshotID, outSnap.ID, ChangeAgentCommit, c.ActorKind, dec.ActorUserIDPtr(), c.RunID, c.ActorSnapshot, diff.Hunks, "", "", fmt.Sprintf("agent_commit:%s:%s", c.RunID, outSnap.ContentSHA256)) + ev := s.putEventLocked(c.DocID, newVer.ID, eventNo, &c.InputSnapshotID, outSnap.ID, ChangeAgentCommit, c.ActorKind, dec.ActorUserIDPtr(), c.RunID, c.ActorSnapshot, diff.Hunks, "", "", fmt.Sprintf("agent_commit:%s:%s", c.RunID, outSnap.ContentSHA256)) c.Status = CandidateCommitted c.DecidedAt = &now + if err := s.persistAgentCommitLocked(ctx, root, oldVer, newVer, outSnap, ev, c); err != nil { + return nil, err + } + if s.Notify && s.OnOutbox != nil { _ = s.OnOutbox(ctx, "document.agent_version_committed", map[string]any{ "doc_id": c.DocID, "version_no": newVer.VersionNo, "risk": rep.RiskLevel, "candidate_id": c.ID, @@ -411,16 +506,17 @@ func (d CommitAgentDecision) ActorUserIDPtr() *int64 { } func (s *Service) ListVersions(ctx context.Context, docID string) ([]Version, error) { - _ = ctx s.mu.Lock() defer s.mu.Unlock() + if err := s.hydrateDocLocked(ctx, docID); err != nil { + return nil, err + } var out []Version for _, v := range s.versions { if v.DocID == docID { out = append(out, *v) } } - // sort by version_no for i := 0; i < len(out); i++ { for j := i + 1; j < len(out); j++ { if out[j].VersionNo < out[i].VersionNo { @@ -432,9 +528,11 @@ func (s *Service) ListVersions(ctx context.Context, docID string) ([]Version, er } func (s *Service) GetVersionContent(ctx context.Context, docID string, versionNo int, view string) (string, string, error) { - _ = ctx s.mu.Lock() defer s.mu.Unlock() + if err := s.hydrateDocLocked(ctx, docID); err != nil { + return "", "", err + } var ver *Version for _, v := range s.versions { if v.DocID == docID && v.VersionNo == versionNo { @@ -469,9 +567,11 @@ func (s *Service) DiffVersions(ctx context.Context, docID string, fromNo, toNo i } func (s *Service) ManualOverlay(ctx context.Context, docID string) (*DiffResult, error) { - _ = ctx s.mu.Lock() defer s.mu.Unlock() + if err := s.hydrateDocLocked(ctx, docID); err != nil { + return nil, err + } d := s.manualOverlayLocked(docID) return &d, nil } @@ -574,10 +674,15 @@ func (s *Service) ProjectOutputMarkdown(docID, repoRoot string) error { snap := s.snapshots[*root.HeadSnapshotID] md := s.blobs[snap.ContentSHA256].ContentMarkdown workDir := root.WorkDir + dataDir := s.DataDir s.mu.Unlock() path := workDir if !filepath.IsAbs(path) { - path = filepath.Join(repoRoot, workDir) + base := repoRoot + if dataDir != "" { + base = dataDir + } + path = filepath.Join(base, workDir) } if err := os.MkdirAll(path, 0o755); err != nil { return err @@ -696,9 +801,10 @@ func (s *Service) manualOverlayLocked(docID string) DiffResult { return d } -func (s *Service) impactLocked(docID string) []DocImpact { - // Deterministic placeholder: no lineage graph in memory mode. - _ = docID +func (s *Service) impactLocked(ctx context.Context, docID string) []DocImpact { + if s.ImpactResolver != nil { + return s.ImpactResolver(ctx, docID) + } return nil }