weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(events): apply version-committed projections only forward

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-15 09:59:55 +00:00
co-authored by Cursor
parent 21d7692c94
commit 92f31ffa6a
5 changed files with 143 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.doc/system-design/module-boundaries.yaml
+7
View File
@@ -62,6 +62,7 @@ imports:
- app/docadvance
- app/docflow
- app/docnotify
- app/docproject
- app/stageflow
- app/uow
- artifactvault
@@ -96,6 +97,9 @@ imports:
- notification
- projectinstance
- projectworkflow
app/docproject:
- docgov
- events
app/docflow:
- app/uow
- docgov
@@ -255,6 +259,8 @@ table_owners:
- events
event_reconciliations:
- events
event_projection_outcomes:
- app/docproject
notifications:
- notification
stage_advance_requests:
@@ -308,6 +314,7 @@ event_producers:
- docrevision
- app/docflow
- events
- app/docproject
document.downstream_impact_detected:
- docrevision
- app/docflow
cmd/events/main.go
+4
View File
@@ -13,6 +13,7 @@ import (
"chat-agent-system/internal/app/docadvance"
"chat-agent-system/internal/app/docnotify"
"chat-agent-system/internal/app/docproject"
"chat-agent-system/internal/config"
"chat-agent-system/internal/db"
"chat-agent-system/internal/docgov"
@@ -77,6 +78,9 @@ func main() {
_ = disp.Register(events.ConsumerWorkflowAdvance, (&docadvance.EventConsumer{
Inst: instSvc, Docs: gov, PW: pwSvc, Notify: notify, Env: cfg.Env,
}).HandleLease)
_ = disp.Register(events.ConsumerProjection, (&docproject.EventConsumer{
Docs: gov, Env: cfg.Env,
}).HandleLease)
mux := http.NewServeMux()
mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"status": "ok", "env": cfg.Env, "owner": owner})
internal/app/docproject/consumer.go
+96
View File
@@ -0,0 +1,96 @@
package docproject
import (
"context"
"encoding/json"
"errors"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/events"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgconn"
)
// EventConsumer applies document.agent_version_committed onto ledger head pointers.
// Closures/producer already project in the approve/commit UoW; this path is idempotent no-replace.
type EventConsumer struct {
Docs DocumentLookup
Env string
}
type DocumentLookup interface {
GetDocument(ctx context.Context, docID string) (*docgov.DocumentRecord, error)
}
func (c *EventConsumer) HandleLease(ctx context.Context, lease events.Lease, tx pgx.Tx) error {
if c == nil {
return events.ErrUnknownConsumer
}
principal := events.ServicePrincipal{
ID: "projection-" + c.Env, Environment: c.Env, Consumer: events.ConsumerProjection,
Actions: []string{"projection.apply"},
}
if err := events.AuthorizePrincipal(principal, lease.Environment, lease.ProjectID, "projection.apply"); err != nil {
return events.ErrQuarantine
}
if lease.EventName != events.EventVersionCommitted {
return nil
}
docID := lease.Envelope.AggregateID
versionNo := int(lease.Envelope.AggregateVersion)
var payload map[string]any
_ = json.Unmarshal(lease.Envelope.Payload, &payload)
if payload != nil {
if id, ok := payload["doc_id"].(string); ok && id != "" {
docID = id
}
if v, ok := payload["version_no"].(float64); ok {
versionNo = int(v)
}
}
if c.Docs == nil || docID == "" {
return events.ErrQuarantine
}
doc, err := c.Docs.GetDocument(ctx, docID)
if err != nil || doc == nil {
return events.ErrQuarantine
}
if doc.ProjectID != lease.ProjectID {
return events.ErrQuarantine
}
var versionID, snapshotID int64
var sha string
err = tx.QueryRow(ctx, `
SELECT v.id, COALESCE(v.head_snapshot_id,0), COALESCE(s.content_sha256,'')
FROM document_versions v
LEFT JOIN document_snapshots s ON s.id = v.head_snapshot_id
WHERE v.doc_id=$1 AND v.version_no=$2
`, docID, versionNo).Scan(&versionID, &snapshotID, &sha)
if err != nil {
return events.ErrQuarantine
}
tag, err := tx.Exec(ctx, `
INSERT INTO event_projection_outcomes (source_event_id, doc_id, version_no, snapshot_id, applied)
VALUES ($1,$2,$3,$4,false)
ON CONFLICT (source_event_id) DO NOTHING
`, lease.EventID, docID, versionNo, snapshotID)
if err != nil {
var pgErr *pgconn.PgError
if errors.As(err, &pgErr) && pgErr.Code == "23505" {
return nil
}
return err
}
if tag.RowsAffected() == 0 {
return nil
}
applied, err := (docgov.Repository{}).ProjectHeadNoReplace(ctx, tx, docID, versionID, snapshotID, versionNo, sha)
if err != nil {
return err
}
_, err = tx.Exec(ctx, `
UPDATE event_projection_outcomes SET applied=$2, snapshot_id=$3 WHERE source_event_id=$1
`, lease.EventID, applied, snapshotID)
return err
}
internal/docgov/repository.go
+22
View File
@@ -6,6 +6,8 @@ import (
"time"
"chat-agent-system/internal/app/uow"
"github.com/jackc/pgx/v5"
)
// Repository is the document_records writer bound to a UnitOfWork scope.
@@ -106,3 +108,23 @@ func (Repository) ProjectHead(ctx context.Context, scope uow.TxScope, docID stri
}
return nil
}
// ProjectHeadNoReplace moves ledger pointers only forward. Equal or newer heads stay.
func (Repository) ProjectHeadNoReplace(ctx context.Context, tx pgx.Tx, docID string, versionID, snapshotID int64, revisionNo int, sha string) (bool, error) {
if tx == nil || docID == "" {
return false, fmt.Errorf("docgov: project head requires tx and doc_id")
}
tag, err := tx.Exec(ctx, `
UPDATE document_records
SET ledger_version_id=$2, head_snapshot_id=$3, head_revision_no=$4,
row_revision=COALESCE(row_revision,0)+1,
content_sha256=COALESCE(NULLIF($5,''), content_sha256),
updated_at=now()
WHERE doc_id=$1
AND COALESCE(head_revision_no, 0) < $4
`, docID, versionID, snapshotID, revisionNo, sha)
if err != nil {
return false, fmt.Errorf("docgov: project head no-replace: %w", err)
}
return tag.RowsAffected() == 1, nil
}
internal/events/registry.go
+14 -14
View File
@@ -12,24 +12,24 @@ const (
ConsumerWorkflowAdvance = "workflow_advance"
ConsumerFAV = "fav"
EventCandidateReady = "document.agent_candidate_ready"
EventVersionCommitted = "document.agent_version_committed"
EventDownstreamImpact = "document.downstream_impact_detected"
EventCandidateReady = "document.agent_candidate_ready"
EventVersionCommitted = "document.agent_version_committed"
EventDownstreamImpact = "document.downstream_impact_detected"
EventChangeReportDegraded = "document.change_report_degraded"
EventApproved = "document.approved"
EventTaskRunCompleted = "taskrun.completed"
EventTaskRunCancelled = "taskrun.cancelled"
EventFAVArtifactStaged = "fav.artifact_staged"
EventApproved = "document.approved"
EventTaskRunCompleted = "taskrun.completed"
EventTaskRunCancelled = "taskrun.cancelled"
EventFAVArtifactStaged = "fav.artifact_staged"
)
// Contract is the static allowlist row for one event name.
type Contract struct {
EventName string
SchemaVersion int
Enabled bool
Producers []string
Consumers []string
PayloadFields []string
EventName string
SchemaVersion int
Enabled bool
Producers []string
Consumers []string
PayloadFields []string
}
// StaticContracts is the runtime copy of module-boundaries event sections.
@@ -37,7 +37,7 @@ type Contract struct {
func StaticContracts() []Contract {
return []Contract{
{EventName: EventCandidateReady, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "candidate_id", "risk"}},
{EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"doc_id", "version_no"}},
{EventName: EventVersionCommitted, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events", "app/docproject"}, Consumers: []string{ConsumerProjection, ConsumerNotify}, PayloadFields: []string{"doc_id", "version_no"}},
{EventName: EventDownstreamImpact, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "impacted_doc_id", "version_no"}},
{EventName: EventChangeReportDegraded, SchemaVersion: 1, Enabled: true, Producers: []string{"docrevision", "app/docflow", "events"}, Consumers: []string{ConsumerNotify}, PayloadFields: []string{"doc_id", "report_id"}},
{EventName: EventApproved, SchemaVersion: 1, Enabled: true, Producers: []string{"app/docflow", "events", "app/docadvance"}, Consumers: []string{ConsumerWorkflowAdvance, ConsumerNotify}, PayloadFields: []string{"doc_id", "project_id", "task_key"}},