|
| 1 | +// Package agentjournal records a handoff as an agent-runtime goal. |
| 2 | +// |
| 3 | +// gds handoff checkpoints unfinished work for a next actor, and that is |
| 4 | +// exactly the contract agent-runtime's Goal journal and handoff lifecycle |
| 5 | +// events were written for: a durable, revisioned, vendor-neutral record of a |
| 6 | +// goal another agent is expected to pick up. The journal lives beside the |
| 7 | +// operation state, the lifecycle events append to a JSONL stream, and both |
| 8 | +// use agent-runtime's own validation -- GDS adds identity and evidence, never |
| 9 | +// a private dialect. |
| 10 | +package agentjournal |
| 11 | + |
| 12 | +import ( |
| 13 | + "context" |
| 14 | + "fmt" |
| 15 | + "os" |
| 16 | + "path/filepath" |
| 17 | + "regexp" |
| 18 | + "strings" |
| 19 | + "time" |
| 20 | + |
| 21 | + "github.com/NDDev-OpenNetwork/agent-runtime/goal" |
| 22 | + "github.com/NDDev-OpenNetwork/agent-runtime/observability" |
| 23 | +) |
| 24 | + |
| 25 | +const ( |
| 26 | + // CheckpointItem is the acceptance item the apply completes: the |
| 27 | + // checkpoint commit exists and is published. |
| 28 | + CheckpointItem = "checkpoint-published" |
| 29 | + // VerifiedItem is the acceptance item the verify completes: the handoff |
| 30 | + // operation re-proved the checkpoint. |
| 31 | + VerifiedItem = "handoff-verified" |
| 32 | +) |
| 33 | + |
| 34 | +var invalidIDRunes = regexp.MustCompile(`[^a-z0-9._-]+`) |
| 35 | + |
| 36 | +// Recorder writes goal journals and lifecycle events under one directory. |
| 37 | +type Recorder struct { |
| 38 | + Directory string |
| 39 | + Now func() time.Time |
| 40 | +} |
| 41 | + |
| 42 | +func (r Recorder) now() time.Time { |
| 43 | + if r.Now != nil { |
| 44 | + return r.Now() |
| 45 | + } |
| 46 | + return time.Now().UTC() |
| 47 | +} |
| 48 | + |
| 49 | +// GoalID derives a valid agent-runtime goal id from an operation id. |
| 50 | +func GoalID(operationID string) string { |
| 51 | + lowered := strings.ToLower(strings.TrimSpace(operationID)) |
| 52 | + lowered = strings.ReplaceAll(lowered, "_", "-") |
| 53 | + lowered = invalidIDRunes.ReplaceAllString(lowered, "-") |
| 54 | + lowered = strings.Trim(lowered, "._-") |
| 55 | + if lowered == "" { |
| 56 | + lowered = "operation" |
| 57 | + } |
| 58 | + return "handoff." + lowered |
| 59 | +} |
| 60 | + |
| 61 | +// JournalPath is where an operation's goal journal lives. |
| 62 | +func (r Recorder) JournalPath(operationID string) string { |
| 63 | + return filepath.Join(r.Directory, GoalID(operationID)+".json") |
| 64 | +} |
| 65 | + |
| 66 | +func (r Recorder) eventsPath() string { |
| 67 | + return filepath.Join(r.Directory, "handoff-events.jsonl") |
| 68 | +} |
| 69 | + |
| 70 | +// RecordCheckpoint creates the goal journal for an applied handoff and marks |
| 71 | +// the checkpoint item complete with its commit evidence, then appends the |
| 72 | +// dispatched handoff event. |
| 73 | +func (r Recorder) RecordCheckpoint( |
| 74 | + ctx context.Context, |
| 75 | + operationID, repositoryID, intent string, |
| 76 | + files []string, |
| 77 | + commitReference string, |
| 78 | + sessionID string, |
| 79 | +) error { |
| 80 | + if err := os.MkdirAll(r.Directory, 0o700); err != nil { |
| 81 | + return fmt.Errorf("create agent journal directory: %w", err) |
| 82 | + } |
| 83 | + now := r.now() |
| 84 | + journal, err := goal.New(GoalID(operationID), intent, []goal.ChecklistItem{ |
| 85 | + {ID: CheckpointItem, Acceptance: "the checkpoint commit exists on the published branch"}, |
| 86 | + {ID: VerifiedItem, Acceptance: "gds handoff --verify re-proved the checkpoint"}, |
| 87 | + }, []string{"integration", "cleanup"}, now) |
| 88 | + if err != nil { |
| 89 | + return fmt.Errorf("draft handoff goal: %w", err) |
| 90 | + } |
| 91 | + evidence := []goal.Evidence{{Type: goal.EvidenceCommit, Reference: commitReference, Result: "checkpoint published for " + repositoryID}} |
| 92 | + for _, file := range files { |
| 93 | + evidence = append(evidence, goal.Evidence{Type: goal.EvidenceFile, Reference: file, Result: "carried by the checkpoint"}) |
| 94 | + } |
| 95 | + store := goal.Store{Path: r.JournalPath(operationID)} |
| 96 | + // The store insists on a genesis create followed by a CAS update -- the |
| 97 | + // same discipline every other consumer gets, so GDS takes it too. |
| 98 | + if err := store.Create(journal); err != nil { |
| 99 | + return fmt.Errorf("store handoff goal: %w", err) |
| 100 | + } |
| 101 | + if _, err := store.Update(journal.Revision, func(stored *goal.Journal) error { |
| 102 | + return stored.CompleteItem(CheckpointItem, evidence, now) |
| 103 | + }); err != nil { |
| 104 | + return fmt.Errorf("complete checkpoint item: %w", err) |
| 105 | + } |
| 106 | + return r.emit(ctx, operationID, sessionID, observability.HandoffStageDispatched) |
| 107 | +} |
| 108 | + |
| 109 | +// RecordVerified marks the verification item complete on the stored journal |
| 110 | +// and appends the completed handoff event. |
| 111 | +func (r Recorder) RecordVerified( |
| 112 | + ctx context.Context, |
| 113 | + operationID string, |
| 114 | + verificationReference string, |
| 115 | + sessionID string, |
| 116 | +) error { |
| 117 | + store := goal.Store{Path: r.JournalPath(operationID)} |
| 118 | + current, err := store.Load() |
| 119 | + if err != nil { |
| 120 | + return fmt.Errorf("load handoff goal: %w", err) |
| 121 | + } |
| 122 | + // A re-verify is idempotent: the journal already says it, and the event |
| 123 | + // stream already carries it, so neither is repeated. |
| 124 | + for _, item := range current.Goal.Acceptance { |
| 125 | + if item.ID == VerifiedItem && item.Status == goal.ItemComplete { |
| 126 | + return nil |
| 127 | + } |
| 128 | + } |
| 129 | + now := r.now() |
| 130 | + if _, err := store.Update(current.Revision, func(journal *goal.Journal) error { |
| 131 | + return journal.CompleteItem(VerifiedItem, []goal.Evidence{{ |
| 132 | + Type: goal.EvidenceCommand, Reference: verificationReference, |
| 133 | + Result: "handoff verify succeeded", |
| 134 | + }}, now) |
| 135 | + }); err != nil { |
| 136 | + return fmt.Errorf("complete verification item: %w", err) |
| 137 | + } |
| 138 | + return r.emit(ctx, operationID, sessionID, observability.HandoffStageCompleted) |
| 139 | +} |
| 140 | + |
| 141 | +func (r Recorder) emit(ctx context.Context, operationID, sessionID string, stage observability.HandoffStage) error { |
| 142 | + sink, err := observability.OpenJSONLSink(r.eventsPath(), observability.JSONLOptions{Name: "gds-handoff"}) |
| 143 | + if err != nil { |
| 144 | + return fmt.Errorf("open handoff event sink: %w", err) |
| 145 | + } |
| 146 | + defer sink.Close(ctx) |
| 147 | + emitter, err := observability.NewEmitter( |
| 148 | + observability.Runtime{ID: "gds", Version: "handoff-v1"}, |
| 149 | + []observability.Sink{sink}, |
| 150 | + observability.Options{Clock: r.now}, |
| 151 | + ) |
| 152 | + if err != nil { |
| 153 | + return fmt.Errorf("build handoff event emitter: %w", err) |
| 154 | + } |
| 155 | + draft, err := observability.HandoffDraft( |
| 156 | + GoalID(operationID), |
| 157 | + observability.ActorWorker, observability.ActorWorker, stage, nil, nil, |
| 158 | + observability.Context{ |
| 159 | + CorrelationID: GoalID(operationID), |
| 160 | + Actor: observability.Actor{Kind: observability.ActorWorker, ID: sessionID}, |
| 161 | + Attempt: observability.AttemptInitial, |
| 162 | + }, |
| 163 | + ) |
| 164 | + if err != nil { |
| 165 | + return fmt.Errorf("draft handoff event: %w", err) |
| 166 | + } |
| 167 | + if _, _, err := emitter.Emit(ctx, draft); err != nil { |
| 168 | + return fmt.Errorf("emit handoff event: %w", err) |
| 169 | + } |
| 170 | + return nil |
| 171 | +} |
0 commit comments