feat(docrevision): persist ledger in PostgreSQL with bootstrap
Hydrate/mutate against schema 00028, ensure first-access bootstrap, and build deterministic change reports instead of static fiction. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
+132
-26
@@ -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
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user