feat(replay): record runtime identity; label compatibility; analyse first divergence
Snapshots, replays and experiments now record the runtime they ran under: build (module version, VCS commit, modified tree, commit time, Go version, platform), schema version, a digest of the non-secret configuration that ignores secret and credential changes, personality version, policy digest, model routing including per-mind overrides, prompt versions, deterministic subsystem versions and timezone. A replay compares its identity with the snapshot's and is an exact-candidate, a different-build (differences listed) or incompatible (schema differs; the replay stops before running). The first non-exact step is analysed into events only in the recording, events only in the replay, and the components they belong to, labelled as inferred; live model calls are named as their own source of difference. Migration 0022 is additive; unrecorded identities are reported as such.
This commit is contained in:
@@ -5,6 +5,8 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
@@ -244,3 +246,19 @@ func (c Config) Redacted() Config {
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// Digest is a stable hash of the non-secret configuration: it changes when
|
||||
// the daemon would behave differently and stays the same when only secrets
|
||||
// or credentials rotate. Used to explain replay differences.
|
||||
func (c Config) Digest() string {
|
||||
public := struct {
|
||||
HTTPAddr, Env, LogLevel, LogFormat, ResearchDir string
|
||||
MaxWorkers, MaxConcurrentLLM int
|
||||
DailyTokenBudget, MaxRequestBody int64
|
||||
SessionIdleTTL, SessionAbsoluteTTL time.Duration
|
||||
HasMasterKey bool
|
||||
}{c.HTTPAddr, c.Env, c.LogLevel, c.LogFormat, c.ResearchDir, c.MaxWorkers, c.MaxConcurrentLLM, c.DailyTokenBudget, c.MaxRequestBody, c.SessionIdleTTL, c.SessionAbsoluteTTL, c.MasterKey != ""}
|
||||
b, _ := json.Marshal(public)
|
||||
sum := sha256.Sum256(b)
|
||||
return "sha256:" + hex.EncodeToString(sum[:8])
|
||||
}
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/andydixon/noema/internal/brain"
|
||||
"github.com/andydixon/noema/internal/identity"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/replay"
|
||||
"github.com/andydixon/noema/internal/snapshots"
|
||||
@@ -68,19 +69,20 @@ type Experiment struct {
|
||||
|
||||
// Results is what a run records.
|
||||
type Results struct {
|
||||
Subject ids.ID `json:"subject_mind_id"`
|
||||
SubjectSlug string `json:"subject_slug"`
|
||||
Steps []StepResult `json:"steps"`
|
||||
EventsByType map[string]int `json:"events_by_type"`
|
||||
Beliefs []BeliefRow `json:"beliefs"`
|
||||
Memories int `json:"memories"`
|
||||
Goals int `json:"goals"`
|
||||
Expectations []ExpectResult `json:"expectations"`
|
||||
LLMCalls map[string]int `json:"llm_calls"` // live / replayed
|
||||
Routing []map[string]any `json:"routing,omitempty"`
|
||||
DurationMS int64 `json:"duration_ms"`
|
||||
Cycles int `json:"cycles"`
|
||||
TruncatedAfter int `json:"truncated_after_stimulus,omitempty"`
|
||||
Subject ids.ID `json:"subject_mind_id"`
|
||||
SubjectSlug string `json:"subject_slug"`
|
||||
Steps []StepResult `json:"steps"`
|
||||
EventsByType map[string]int `json:"events_by_type"`
|
||||
Beliefs []BeliefRow `json:"beliefs"`
|
||||
Memories int `json:"memories"`
|
||||
Goals int `json:"goals"`
|
||||
Expectations []ExpectResult `json:"expectations"`
|
||||
LLMCalls map[string]int `json:"llm_calls"` // live / replayed
|
||||
Routing []map[string]any `json:"routing,omitempty"`
|
||||
DurationMS int64 `json:"duration_ms"`
|
||||
Identity identity.Identity `json:"identity"` // the runtime that produced these results
|
||||
Cycles int `json:"cycles"`
|
||||
TruncatedAfter int `json:"truncated_after_stimulus,omitempty"`
|
||||
}
|
||||
|
||||
// StepResult records one stimulus.
|
||||
@@ -114,6 +116,7 @@ type Service struct {
|
||||
Snapshots *snapshots.Store
|
||||
Brain brain.Config
|
||||
Routing func(ctx context.Context) []map[string]any // current LLM role routing, recorded with results
|
||||
Identity *identity.Capturer
|
||||
Log *slog.Logger
|
||||
}
|
||||
|
||||
@@ -219,6 +222,7 @@ func (s *Service) execute(ctx context.Context, e *Experiment, by string) (Result
|
||||
if s.Routing != nil {
|
||||
res.Routing = s.Routing(ctx)
|
||||
}
|
||||
res.Identity = s.Identity.Capture(ctx, subject.ID)
|
||||
drv, err := replay.NewDriver(ctx, s.Brain, subject, "experiment")
|
||||
if err != nil {
|
||||
return res, err
|
||||
|
||||
@@ -0,0 +1,229 @@
|
||||
// Package identity records enough about the running system to explain, later,
|
||||
// why two runs of the same cognition differed: the build, the schema, the
|
||||
// non-secret configuration, the model routing, prompt and personality
|
||||
// versions and the policy in force. Replays and experiments store it and
|
||||
// compare it; a difference is reported, never papered over.
|
||||
package identity
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
"github.com/andydixon/noema/internal/capabilities"
|
||||
"github.com/andydixon/noema/internal/config"
|
||||
"github.com/andydixon/noema/internal/llm/prompts"
|
||||
"github.com/andydixon/noema/internal/llm/router"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
"github.com/andydixon/noema/internal/version"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
)
|
||||
|
||||
// Identity is the runtime identity captured with a snapshot, replay or
|
||||
// experiment. Everything in it is safe to display: no secret or secret
|
||||
// reference is ever included.
|
||||
type Identity struct {
|
||||
CapturedAt time.Time `json:"captured_at"`
|
||||
Build version.Build `json:"build"`
|
||||
SchemaVersion int `json:"schema_version"`
|
||||
ConfigDigest string `json:"config_digest"`
|
||||
Timezone string `json:"timezone"`
|
||||
PersonalityVersion int `json:"personality_version,omitempty"`
|
||||
PolicyDigest string `json:"policy_digest,omitempty"`
|
||||
Routing map[string]string `json:"routing,omitempty"` // role → provider/model
|
||||
PromptVersions map[string]int `json:"prompt_versions,omitempty"` // prompt name → current version
|
||||
Deterministic map[string]string `json:"deterministic"` // versions of deterministic subsystems whose output is compared
|
||||
}
|
||||
|
||||
// Deterministic subsystem versions. Bump one when its output for the same
|
||||
// input changes, so a replay can name the component that moved.
|
||||
var Deterministic = map[string]string{
|
||||
"attention": "1",
|
||||
"retrieval": "1",
|
||||
"perception": "1",
|
||||
"epistemics": "1",
|
||||
"confidence": "2", // family grouping
|
||||
"contradiction": "2", // temporal supersession
|
||||
"association": "2", // speech acts excluded
|
||||
}
|
||||
|
||||
// Capturer gathers an identity from the live system. Every field is optional
|
||||
// so tests and partial deployments still get a truthful, smaller record.
|
||||
type Capturer struct {
|
||||
DB *pgxpool.Pool
|
||||
Config *config.Config
|
||||
Minds *minds.Store
|
||||
LLM *router.Store
|
||||
Prompts *prompts.Store
|
||||
Capabilities *capabilities.Service
|
||||
}
|
||||
|
||||
// Capture records the identity for a mind (ids.Nil for daemon-wide).
|
||||
func (c *Capturer) Capture(ctx context.Context, mind ids.ID) Identity {
|
||||
id := Identity{CapturedAt: time.Now().UTC(), Build: version.Info(), Timezone: time.Now().Location().String(), Deterministic: Deterministic}
|
||||
if c == nil {
|
||||
return id
|
||||
}
|
||||
if c.DB != nil {
|
||||
id.SchemaVersion, _ = storage.SchemaVersion(ctx, c.DB)
|
||||
}
|
||||
if c.Config != nil {
|
||||
id.ConfigDigest = c.Config.Digest()
|
||||
}
|
||||
if c.Minds != nil && !mind.IsZero() {
|
||||
if m, err := c.Minds.Get(ctx, mind); err == nil {
|
||||
id.PersonalityVersion = m.PersonalityVersion
|
||||
}
|
||||
}
|
||||
if c.LLM != nil {
|
||||
id.Routing = map[string]string{}
|
||||
models := map[ids.ID]router.Model{}
|
||||
if ms, err := c.LLM.ListModels(ctx); err == nil {
|
||||
for _, m := range ms {
|
||||
models[m.ID] = m
|
||||
}
|
||||
}
|
||||
if roles, err := c.LLM.ListRoles(ctx); err == nil {
|
||||
for _, r := range roles {
|
||||
if r.ModelID != nil {
|
||||
if m, ok := models[*r.ModelID]; ok {
|
||||
id.Routing[string(r.Role)] = m.Provider + "/" + m.Name
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if c.Minds != nil && !mind.IsZero() {
|
||||
if m, err := c.Minds.Get(ctx, mind); err == nil {
|
||||
for role, mid := range m.Config.LLMRoles {
|
||||
if md, ok := models[ids.ID(mid)]; ok {
|
||||
id.Routing[role] = md.Provider + "/" + md.Name + " (mind override)"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if c.Prompts != nil {
|
||||
if ps, err := c.Prompts.List(ctx); err == nil {
|
||||
id.PromptVersions = map[string]int{}
|
||||
for _, p := range ps {
|
||||
id.PromptVersions[p.Name] = p.CurrentVersion
|
||||
}
|
||||
}
|
||||
}
|
||||
if c.Capabilities != nil && !mind.IsZero() {
|
||||
if ps, err := c.Capabilities.Policies(ctx, mind); err == nil {
|
||||
id.PolicyDigest = policyDigest(ps)
|
||||
}
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func policyDigest(ps []capabilities.Policy) string {
|
||||
type row struct {
|
||||
Mind, Capability, Target, Action, Effect string
|
||||
From, To *int
|
||||
}
|
||||
var rows []row
|
||||
for _, p := range ps {
|
||||
rows = append(rows, row{p.MindID.String(), p.Capability, p.TargetPattern, p.ActionKind, string(p.Effect), p.HoursFrom, p.HoursTo})
|
||||
}
|
||||
sort.Slice(rows, func(i, j int) bool { return fmt.Sprint(rows[i]) < fmt.Sprint(rows[j]) })
|
||||
b, _ := json.Marshal(rows)
|
||||
sum := sha256.Sum256(b)
|
||||
return "sha256:" + hex.EncodeToString(sum[:8])
|
||||
}
|
||||
|
||||
// Compatibility labels for comparing the identity a snapshot was taken under
|
||||
// with the identity a replay runs under.
|
||||
const (
|
||||
CompatExactCandidate = "exact-candidate" // same schema and build; differences can only come from state or models
|
||||
CompatDifferentBuild = "different-build" // same schema, other code: exact steps are still exact, but divergence may be code
|
||||
CompatIncompatible = "incompatible" // schema differs: the sandbox cannot be a faithful reconstruction
|
||||
)
|
||||
|
||||
// Compare lists the fields that differ and the resulting compatibility.
|
||||
func Compare(recorded, current Identity) (string, []string) {
|
||||
var diffs []string
|
||||
if recorded.Build == (version.Build{}) && recorded.SchemaVersion == 0 {
|
||||
// Taken before identities were recorded: say so; do not compare blanks.
|
||||
return CompatDifferentBuild, []string{"snapshot identity not recorded (taken before 0.2); the comparison cannot rule out code differences"}
|
||||
}
|
||||
if recorded.SchemaVersion != 0 && recorded.SchemaVersion != current.SchemaVersion {
|
||||
diffs = append(diffs, fmt.Sprintf("schema %d → %d", recorded.SchemaVersion, current.SchemaVersion))
|
||||
}
|
||||
sameBuild := recorded.Build == current.Build
|
||||
if !sameBuild {
|
||||
diffs = append(diffs, fmt.Sprintf("build %s%s → %s%s", short(recorded.Build.Commit, recorded.Build.Version), dirty(recorded.Build.Modified), short(current.Build.Commit, current.Build.Version), dirty(current.Build.Modified)))
|
||||
}
|
||||
if recorded.ConfigDigest != "" && recorded.ConfigDigest != current.ConfigDigest {
|
||||
diffs = append(diffs, "configuration digest differs")
|
||||
}
|
||||
if recorded.PersonalityVersion != 0 && recorded.PersonalityVersion != current.PersonalityVersion {
|
||||
diffs = append(diffs, fmt.Sprintf("personality v%d → v%d", recorded.PersonalityVersion, current.PersonalityVersion))
|
||||
}
|
||||
if recorded.PolicyDigest != "" && recorded.PolicyDigest != current.PolicyDigest {
|
||||
diffs = append(diffs, "policy differs")
|
||||
}
|
||||
for role, m := range recorded.Routing {
|
||||
if cur := current.Routing[role]; cur != m {
|
||||
diffs = append(diffs, fmt.Sprintf("routing %s: %s → %s", role, m, orNone(cur)))
|
||||
}
|
||||
}
|
||||
for name, v := range recorded.PromptVersions {
|
||||
if cur := current.PromptVersions[name]; cur != v {
|
||||
diffs = append(diffs, fmt.Sprintf("prompt %s: v%d → v%d", name, v, cur))
|
||||
}
|
||||
}
|
||||
for comp, v := range recorded.Deterministic {
|
||||
if cur := current.Deterministic[comp]; cur != v {
|
||||
diffs = append(diffs, fmt.Sprintf("%s subsystem v%s → v%s", comp, v, orNone(cur)))
|
||||
}
|
||||
}
|
||||
if recorded.Timezone != "" && recorded.Timezone != current.Timezone {
|
||||
diffs = append(diffs, "timezone "+recorded.Timezone+" → "+current.Timezone)
|
||||
}
|
||||
sort.Strings(diffs)
|
||||
switch {
|
||||
case recorded.SchemaVersion != 0 && recorded.SchemaVersion != current.SchemaVersion:
|
||||
return CompatIncompatible, diffs
|
||||
case len(diffs) > 0:
|
||||
return CompatDifferentBuild, diffs
|
||||
}
|
||||
if current.Build.Modified {
|
||||
// Not a difference between the two runs, but worth saying: the commit
|
||||
// alone does not describe the code either of them ran.
|
||||
diffs = append(diffs, "note: built from a modified working tree")
|
||||
}
|
||||
return CompatExactCandidate, diffs
|
||||
}
|
||||
|
||||
func short(commit, v string) string {
|
||||
if len(commit) >= 8 {
|
||||
return commit[:8]
|
||||
}
|
||||
if commit != "" {
|
||||
return commit
|
||||
}
|
||||
return v
|
||||
}
|
||||
|
||||
func dirty(m bool) string {
|
||||
if m {
|
||||
return "+dirty"
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func orNone(s string) string {
|
||||
if s == "" {
|
||||
return "none"
|
||||
}
|
||||
return s
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package identity
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/andydixon/noema/internal/config"
|
||||
"github.com/andydixon/noema/internal/version"
|
||||
)
|
||||
|
||||
func TestConfigDigestIgnoresSecrets(t *testing.T) {
|
||||
a := config.Defaults()
|
||||
a.MasterKey, a.DatabaseURL, a.BootstrapAdminPassword = "k1", "postgres://u:p1@h/db", "pw1"
|
||||
b := a
|
||||
b.MasterKey, b.DatabaseURL, b.BootstrapAdminPassword = "k2", "postgres://u:p2@other/db", "pw2"
|
||||
if a.Digest() != b.Digest() {
|
||||
t.Fatal("digest must not change when only secrets or credentials change")
|
||||
}
|
||||
c := a
|
||||
c.MaxWorkers++
|
||||
if c.Digest() == a.Digest() {
|
||||
t.Fatal("digest must change when behaviour-affecting configuration changes")
|
||||
}
|
||||
for _, d := range []string{a.Digest(), b.Digest()} {
|
||||
for _, secret := range []string{"k1", "k2", "p1", "p2", "pw1", "pw2"} {
|
||||
if len(secret) > 0 && contains(d, secret) {
|
||||
t.Fatalf("digest leaks %q", secret)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCompareLabels(t *testing.T) {
|
||||
base := Identity{SchemaVersion: 22, Build: version.Build{Commit: "abcdef1234567890", GoVersion: "go1.26"}, ConfigDigest: "sha256:x", Routing: map[string]string{"respond": "openai/gpt"}, PromptVersions: map[string]int{"respond": 1}, Deterministic: Deterministic}
|
||||
if c, d := Compare(base, base); c != CompatExactCandidate || len(d) != 0 {
|
||||
t.Fatalf("identical: %s %v", c, d)
|
||||
}
|
||||
other := base
|
||||
other.Build = version.Build{Commit: "0123456789abcdef", GoVersion: "go1.26"}
|
||||
if c, d := Compare(base, other); c != CompatDifferentBuild || len(d) != 1 {
|
||||
t.Fatalf("different build: %s %v", c, d)
|
||||
}
|
||||
rerouted := base
|
||||
rerouted.Routing = map[string]string{"respond": "anthropic/claude"}
|
||||
if c, d := Compare(base, rerouted); c != CompatDifferentBuild || len(d) != 1 {
|
||||
t.Fatalf("rerouted: %s %v", c, d)
|
||||
}
|
||||
older := base
|
||||
older.SchemaVersion = 21
|
||||
if c, _ := Compare(older, base); c != CompatIncompatible {
|
||||
t.Fatalf("schema change must be incompatible: %s", c)
|
||||
}
|
||||
// An unrecorded identity (pre-0.2 snapshot) is named as such rather than
|
||||
// compared blank against blank or treated as an exact candidate.
|
||||
if c, d := Compare(Identity{}, base); c != CompatDifferentBuild || len(d) != 1 {
|
||||
t.Fatalf("unrecorded: %s %v", c, d)
|
||||
}
|
||||
}
|
||||
|
||||
func contains(s, sub string) bool {
|
||||
return len(sub) > 0 && len(s) >= len(sub) && (s == sub || indexOf(s, sub) >= 0)
|
||||
}
|
||||
|
||||
func indexOf(s, sub string) int {
|
||||
for i := 0; i+len(sub) <= len(s); i++ {
|
||||
if s[i:i+len(sub)] == sub {
|
||||
return i
|
||||
}
|
||||
}
|
||||
return -1
|
||||
}
|
||||
@@ -42,6 +42,11 @@ type CognitiveConfig struct {
|
||||
DreamLLMCalls int `json:"dream_llm_calls"`
|
||||
DreamMaxMemories int `json:"dream_max_memories"`
|
||||
DreamMaxHypotheses int `json:"dream_max_hypotheses"`
|
||||
|
||||
// LLMRoles overrides the daemon-wide model routing for this mind:
|
||||
// role → model id. It is what lets two forks of one snapshot run on
|
||||
// different providers while sharing everything else.
|
||||
LLMRoles map[string]string `json:"llm_roles,omitempty"`
|
||||
}
|
||||
|
||||
// AttentionFeatures is the fixed set of scoring features (spec §6). Weights are
|
||||
|
||||
+145
-6
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -13,6 +14,7 @@ import (
|
||||
|
||||
"github.com/andydixon/noema/internal/brain"
|
||||
evstore "github.com/andydixon/noema/internal/events"
|
||||
"github.com/andydixon/noema/internal/identity"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/snapshots"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
@@ -35,6 +37,27 @@ type Replay struct {
|
||||
CreatedBy string `json:"created_by"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
FinishedAt *time.Time `json:"finished_at,omitempty"`
|
||||
// Reproducibility: the runtime this replay ran under, how it compares
|
||||
// with the runtime that took the snapshot, and where the first
|
||||
// divergence appeared.
|
||||
Identity identity.Identity `json:"identity"`
|
||||
Compatibility string `json:"compatibility"` // exact-candidate | different-build | incompatible
|
||||
IdentityDiff []string `json:"identity_diff"`
|
||||
Divergence *Divergence `json:"divergence,omitempty"`
|
||||
}
|
||||
|
||||
// Divergence describes the first step whose outcome differed and which
|
||||
// components the differing events belong to. Components are inferred from
|
||||
// event types and labelled as such; they are a pointer, not a proof.
|
||||
type Divergence struct {
|
||||
Step int `json:"step"`
|
||||
StimulusID ids.ID `json:"stimulus_id"`
|
||||
Label string `json:"label"`
|
||||
Components []string `json:"components"`
|
||||
OnlyRecorded []string `json:"only_recorded"`
|
||||
OnlyReplayed []string `json:"only_replayed"`
|
||||
RecordedIn string `json:"recorded_build"`
|
||||
ReplayedIn string `json:"replayed_build"`
|
||||
}
|
||||
|
||||
// Summary counts step labels.
|
||||
@@ -70,6 +93,7 @@ type Service struct {
|
||||
Minds *minds.Store
|
||||
Snapshots *snapshots.Store
|
||||
Brain brain.Config // template; Services.LLM must be a replaying router
|
||||
Identity *identity.Capturer
|
||||
Log *slog.Logger
|
||||
}
|
||||
|
||||
@@ -105,7 +129,13 @@ func (s *Service) Begin(ctx context.Context, src minds.Mind, snapshotID ids.ID,
|
||||
if r.ToSeq <= r.FromSeq {
|
||||
return r, fmt.Errorf("nothing to replay after event %d", r.FromSeq)
|
||||
}
|
||||
_, err = s.DB.Exec(ctx, `INSERT INTO replays (id, mind_id, snapshot_id, from_seq, to_seq, status, created_by) VALUES ($1,$2,$3,$4,$5,'running',$6)`, r.ID, src.ID, snapshotID, r.FromSeq, r.ToSeq, by)
|
||||
r.Identity = s.Identity.Capture(ctx, src.ID)
|
||||
r.Compatibility, r.IdentityDiff = identity.Compare(sn.Identity, r.Identity)
|
||||
if r.IdentityDiff == nil {
|
||||
r.IdentityDiff = []string{}
|
||||
}
|
||||
idj, _ := json.Marshal(r.Identity)
|
||||
_, err = s.DB.Exec(ctx, `INSERT INTO replays (id, mind_id, snapshot_id, from_seq, to_seq, status, created_by, identity, compatibility, identity_diff) VALUES ($1,$2,$3,$4,$5,'running',$6,$7,$8,$9)`, r.ID, src.ID, snapshotID, r.FromSeq, r.ToSeq, by, idj, r.Compatibility, r.IdentityDiff)
|
||||
return r, err
|
||||
}
|
||||
|
||||
@@ -128,7 +158,11 @@ func (s *Service) Execute(ctx context.Context, r Replay) (Replay, error) {
|
||||
}
|
||||
sj, _ := json.Marshal(steps)
|
||||
sum, _ := json.Marshal(r.Summary)
|
||||
_, err := s.DB.Exec(ctx, `UPDATE replays SET status=$2, error=$3, steps=$4, summary=$5, replay_mind_id=$6, finished_at=now() WHERE id=$1`, r.ID, status, errText, sj, sum, r.ReplayMindID)
|
||||
div := []byte("{}")
|
||||
if r.Divergence != nil {
|
||||
div, _ = json.Marshal(r.Divergence)
|
||||
}
|
||||
_, err := s.DB.Exec(ctx, `UPDATE replays SET status=$2, error=$3, steps=$4, summary=$5, replay_mind_id=$6, divergence=$7, finished_at=now() WHERE id=$1`, r.ID, status, errText, sj, sum, r.ReplayMindID, div)
|
||||
r.Status, r.Error, r.Steps = status, errText, sj
|
||||
if err != nil {
|
||||
return r, err
|
||||
@@ -142,6 +176,11 @@ func (s *Service) Execute(ctx context.Context, r Replay) (Replay, error) {
|
||||
if err != nil || sn.MindID != src.ID {
|
||||
return finish("failed", "snapshot not found for this mind", nil)
|
||||
}
|
||||
if r.Compatibility == identity.CompatIncompatible {
|
||||
// A different schema means the sandbox could not be a faithful
|
||||
// reconstruction; say so instead of producing a misleading comparison.
|
||||
return finish("incompatible", "schema differs from the one the snapshot was taken under: "+strings.Join(r.IdentityDiff, "; "), nil)
|
||||
}
|
||||
stimuli, err := s.Events.Query(ctx, evstore.Filter{MindID: src.ID, Categories: StimulusCategories, AfterSeq: from, BeforeSeq: to + 1, Ascending: true, Limit: MaxStimuli + 1})
|
||||
if err != nil {
|
||||
return finish("failed", err.Error(), nil)
|
||||
@@ -211,6 +250,9 @@ func (s *Service) Execute(ctx context.Context, r Replay) (Replay, error) {
|
||||
r.Summary.Diverged++
|
||||
}
|
||||
steps = append(steps, st)
|
||||
if st.Label != "exact" && r.Divergence == nil {
|
||||
r.Divergence = analyseDivergence(st, sn.Identity, r.Identity)
|
||||
}
|
||||
}
|
||||
r.Summary.Steps = len(steps)
|
||||
return finish("done", "", steps)
|
||||
@@ -247,15 +289,25 @@ func (s *Service) llmCounts(ctx context.Context, mind ids.ID, since time.Time) (
|
||||
return
|
||||
}
|
||||
|
||||
const replayCols = `id, mind_id, snapshot_id, replay_mind_id, from_seq, to_seq, status, summary, steps, error, created_by, created_at, finished_at`
|
||||
const replayCols = `id, mind_id, snapshot_id, replay_mind_id, from_seq, to_seq, status, summary, steps, error, created_by, created_at, finished_at, identity, compatibility, identity_diff, divergence`
|
||||
|
||||
func scan(row pgx.Row) (Replay, error) {
|
||||
var r Replay
|
||||
var sum []byte
|
||||
if err := row.Scan(&r.ID, &r.MindID, &r.SnapshotID, &r.ReplayMindID, &r.FromSeq, &r.ToSeq, &r.Status, &sum, &r.Steps, &r.Error, &r.CreatedBy, &r.CreatedAt, &r.FinishedAt); err != nil {
|
||||
var sum, idj, div []byte
|
||||
if err := row.Scan(&r.ID, &r.MindID, &r.SnapshotID, &r.ReplayMindID, &r.FromSeq, &r.ToSeq, &r.Status, &sum, &r.Steps, &r.Error, &r.CreatedBy, &r.CreatedAt, &r.FinishedAt, &idj, &r.Compatibility, &r.IdentityDiff, &div); err != nil {
|
||||
return r, storage.Translate(err)
|
||||
}
|
||||
json.Unmarshal(sum, &r.Summary) //nolint:errcheck
|
||||
json.Unmarshal(sum, &r.Summary) //nolint:errcheck
|
||||
json.Unmarshal(idj, &r.Identity) //nolint:errcheck
|
||||
if len(div) > 2 {
|
||||
var d Divergence
|
||||
if json.Unmarshal(div, &d) == nil {
|
||||
r.Divergence = &d
|
||||
}
|
||||
}
|
||||
if r.IdentityDiff == nil {
|
||||
r.IdentityDiff = []string{}
|
||||
}
|
||||
return r, nil
|
||||
}
|
||||
|
||||
@@ -291,3 +343,90 @@ func StepsOf(r Replay) []Step {
|
||||
json.Unmarshal(r.Steps, &out) //nolint:errcheck
|
||||
return out
|
||||
}
|
||||
|
||||
// componentOf maps an event type onto the deterministic subsystem that
|
||||
// produces it, so a divergence can point at a component.
|
||||
func componentOf(digest string) string {
|
||||
t, _, _ := strings.Cut(digest, ":")
|
||||
switch t {
|
||||
case "memory_retrieved":
|
||||
return "retrieval"
|
||||
case "thought_selected", "thought_candidate", "attention_shift":
|
||||
return "attention"
|
||||
case "memory_created", "memory_reinforced":
|
||||
return "encoding"
|
||||
case "belief_created", "belief_updated", "belief_challenged", "belief_retracted", "contradiction_detected":
|
||||
return "beliefs"
|
||||
case "hypothesis_created", "hypothesis_supported", "hypothesis_weakened":
|
||||
return "association"
|
||||
case "prediction_created", "prediction_resolved", "prediction_error":
|
||||
return "predictions"
|
||||
case "goal_created", "goal_updated", "goal_completed", "goal_abandoned":
|
||||
return "goals"
|
||||
case "response_generated":
|
||||
return "response"
|
||||
case "narrative_updated":
|
||||
return "narrative"
|
||||
case "drive_changed", "emotion_changed":
|
||||
return "motivation"
|
||||
case "metacognition":
|
||||
return "metacognition"
|
||||
case "tool_requested", "tool_denied", "policy_decision":
|
||||
return "capabilities"
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
// analyseDivergence records the first non-exact step: what only the
|
||||
// recording had, what only the replay had, and the components involved.
|
||||
func analyseDivergence(st Step, recorded, current identity.Identity) *Divergence {
|
||||
d := &Divergence{Step: st.Index, StimulusID: st.StimulusID, Label: st.Label, RecordedIn: shortBuild(recorded), ReplayedIn: shortBuild(current)}
|
||||
have := map[string]bool{}
|
||||
for _, x := range st.Replayed {
|
||||
have[x] = true
|
||||
}
|
||||
comps := map[string]bool{}
|
||||
for _, x := range st.Original {
|
||||
if !have[x] {
|
||||
d.OnlyRecorded = append(d.OnlyRecorded, x)
|
||||
comps[componentOf(x)] = true
|
||||
}
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, x := range st.Original {
|
||||
seen[x] = true
|
||||
}
|
||||
for _, x := range st.Replayed {
|
||||
if !seen[x] {
|
||||
d.OnlyReplayed = append(d.OnlyReplayed, x)
|
||||
comps[componentOf(x)] = true
|
||||
}
|
||||
}
|
||||
for c := range comps {
|
||||
d.Components = append(d.Components, c)
|
||||
}
|
||||
sort.Strings(d.Components)
|
||||
if st.LLMLive > 0 {
|
||||
d.Components = append(d.Components, "model (live call; provider nondeterminism)")
|
||||
}
|
||||
if d.OnlyRecorded == nil {
|
||||
d.OnlyRecorded = []string{}
|
||||
}
|
||||
if d.OnlyReplayed == nil {
|
||||
d.OnlyReplayed = []string{}
|
||||
}
|
||||
return d
|
||||
}
|
||||
|
||||
func shortBuild(id identity.Identity) string {
|
||||
if len(id.Build.Commit) >= 8 {
|
||||
if id.Build.Modified {
|
||||
return id.Build.Commit[:8] + "+dirty"
|
||||
}
|
||||
return id.Build.Commit[:8]
|
||||
}
|
||||
if id.Build.Version != "" {
|
||||
return id.Build.Version
|
||||
}
|
||||
return "unrecorded"
|
||||
}
|
||||
|
||||
@@ -2,12 +2,14 @@ package replay
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"log/slog"
|
||||
"sort"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/andydixon/noema/internal/beliefs"
|
||||
"github.com/andydixon/noema/internal/identity"
|
||||
|
||||
evstore "github.com/andydixon/noema/internal/events"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
@@ -40,7 +42,7 @@ func TestReplayIsExactWithoutModel(t *testing.T) {
|
||||
t.Fatalf("live run produced nothing: %v cycles=%d", err, cycles)
|
||||
}
|
||||
last, _ := h.Events.Query(ctx, evstore.Filter{MindID: src.ID, Limit: 1})
|
||||
svc := &Service{DB: h.Pool, Events: h.Events, Minds: h.Minds, Snapshots: h.Snaps, Brain: h.Brain, Log: slog.New(slog.DiscardHandler)}
|
||||
svc := &Service{DB: h.Pool, Events: h.Events, Minds: h.Minds, Snapshots: h.Snaps, Brain: h.Brain, Identity: h.Identity, Log: slog.New(slog.DiscardHandler)}
|
||||
r, err := svc.Run(ctx, src, sn.ID, sn.EventSeq, last[0].Seq, "test")
|
||||
if err != nil {
|
||||
t.Fatalf("replay: %v %+v", err, r)
|
||||
@@ -48,6 +50,25 @@ func TestReplayIsExactWithoutModel(t *testing.T) {
|
||||
if r.Status != "done" || r.Summary.Steps != 1 || r.Summary.Exact != 1 || r.Summary.LLMLive != 0 {
|
||||
t.Fatalf("expected one exact step: %+v steps=%s", r.Summary, string(r.Steps))
|
||||
}
|
||||
// Reproducibility record: the runtime is recorded and compared with the
|
||||
// snapshot's; same process, same schema means an exact candidate with no
|
||||
// first divergence.
|
||||
if r.Identity.SchemaVersion == 0 || r.Identity.Build.GoVersion == "" || len(r.Identity.Deterministic) == 0 {
|
||||
t.Fatalf("identity not recorded: %+v", r.Identity)
|
||||
}
|
||||
if r.Compatibility != identity.CompatExactCandidate || r.Divergence != nil {
|
||||
t.Fatalf("compatibility %q diff %v divergence %+v", r.Compatibility, r.IdentityDiff, r.Divergence)
|
||||
}
|
||||
// A snapshot taken under another schema cannot be replayed faithfully.
|
||||
old, _ := h.Snaps.Get(ctx, sn.ID)
|
||||
old.Identity.SchemaVersion = 1
|
||||
oj, _ := json.Marshal(old.Identity)
|
||||
h.Pool.Exec(ctx, `UPDATE snapshots SET identity=$2 WHERE id=$1`, sn.ID, oj) //nolint:errcheck
|
||||
inc, err := svc.Run(ctx, src, sn.ID, sn.EventSeq, last[0].Seq, "test")
|
||||
if inc.Status != "incompatible" || inc.Compatibility != identity.CompatIncompatible || inc.Summary.Steps != 0 {
|
||||
t.Fatalf("incompatible replay should stop before running: %+v %v", inc, err)
|
||||
}
|
||||
h.Pool.Exec(ctx, `UPDATE snapshots SET identity=$2 WHERE id=$1`, sn.ID, mustJSON(r.Identity)) //nolint:errcheck
|
||||
steps := StepsOf(r)
|
||||
if len(steps) != 1 || !Same(steps[0].Original, steps[0].Replayed) || len(steps[0].Replayed) == 0 {
|
||||
t.Fatalf("step mismatch: %+v", steps)
|
||||
@@ -96,3 +117,8 @@ func provenanceDigest(t *testing.T, h *replaytest.Harness, mind ids.ID) string {
|
||||
sort.Strings(lines)
|
||||
return strings.Join(lines, "\n")
|
||||
}
|
||||
|
||||
func mustJSON(v any) []byte {
|
||||
b, _ := json.Marshal(v)
|
||||
return b
|
||||
}
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
evstore "github.com/andydixon/noema/internal/events"
|
||||
"github.com/andydixon/noema/internal/goals"
|
||||
"github.com/andydixon/noema/internal/graph"
|
||||
"github.com/andydixon/noema/internal/identity"
|
||||
"github.com/andydixon/noema/internal/memory"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/motivation"
|
||||
@@ -31,11 +32,12 @@ import (
|
||||
|
||||
// Harness wires the deterministic cognitive registry without a model.
|
||||
type Harness struct {
|
||||
Pool *pgxpool.Pool
|
||||
Events *evstore.Store
|
||||
Minds *minds.Store
|
||||
Snaps *snapshots.Store
|
||||
Brain brain.Config
|
||||
Pool *pgxpool.Pool
|
||||
Events *evstore.Store
|
||||
Minds *minds.Store
|
||||
Snaps *snapshots.Store
|
||||
Brain brain.Config
|
||||
Identity *identity.Capturer
|
||||
}
|
||||
|
||||
// NewHarness is shared with the experiments tests.
|
||||
@@ -57,5 +59,6 @@ func NewHarness(t *testing.T) *Harness {
|
||||
cognition.Handlers{Memory: mem, Beliefs: bs, Conversations: conversation.NewStore(pool), Graph: graph.NewStore(pool), Assertions: epistemics.NewStore(pool)}.Register(reg)
|
||||
cfg := brain.Config{DB: pool, Events: es, Minds: ms, Registry: reg, Pool: scheduler.NewPool(2), Log: log, Metrics: observability.NewMetrics(),
|
||||
Services: brain.Services{Events: es, Memory: mem, Beliefs: bs, Goals: goals.NewStore(pool), Motivation: motivation.NewStore(pool), Predictions: predictions.NewStore(pool), Capabilities: caps}}
|
||||
return &Harness{Pool: pool, Events: es, Minds: ms, Snaps: &snapshots.Store{DB: pool, Events: es, Minds: ms}, Brain: cfg}
|
||||
ident := &identity.Capturer{DB: pool, Minds: ms}
|
||||
return &Harness{Pool: pool, Events: es, Minds: ms, Snaps: &snapshots.Store{DB: pool, Events: es, Minds: ms, Identity: ident}, Brain: cfg, Identity: ident}
|
||||
}
|
||||
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
|
||||
evstore "github.com/andydixon/noema/internal/events"
|
||||
"github.com/andydixon/noema/internal/identity"
|
||||
"github.com/andydixon/noema/internal/minds"
|
||||
"github.com/andydixon/noema/internal/storage"
|
||||
"github.com/andydixon/noema/pkg/ids"
|
||||
@@ -335,23 +336,25 @@ func remapValue(v any, remap map[string]string) any {
|
||||
|
||||
// Snapshot is a stored archive.
|
||||
type Snapshot struct {
|
||||
ID ids.ID `json:"id"`
|
||||
MindID ids.ID `json:"mind_id"`
|
||||
Name string `json:"name"`
|
||||
Note string `json:"note"`
|
||||
EventSeq int64 `json:"event_seq"`
|
||||
Checksum string `json:"checksum"`
|
||||
SizeBytes int `json:"size_bytes"`
|
||||
Counts map[string]int `json:"counts"`
|
||||
CreatedBy string `json:"created_by"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
ID ids.ID `json:"id"`
|
||||
MindID ids.ID `json:"mind_id"`
|
||||
Name string `json:"name"`
|
||||
Note string `json:"note"`
|
||||
EventSeq int64 `json:"event_seq"`
|
||||
Checksum string `json:"checksum"`
|
||||
SizeBytes int `json:"size_bytes"`
|
||||
Counts map[string]int `json:"counts"`
|
||||
CreatedBy string `json:"created_by"`
|
||||
CreatedAt time.Time `json:"created_at"`
|
||||
Identity identity.Identity `json:"identity"` // the runtime that captured it
|
||||
}
|
||||
|
||||
// Store persists snapshots and performs forks and imports.
|
||||
type Store struct {
|
||||
DB *pgxpool.Pool
|
||||
Events *evstore.Store
|
||||
Minds *minds.Store
|
||||
DB *pgxpool.Pool
|
||||
Events *evstore.Store
|
||||
Minds *minds.Store
|
||||
Identity *identity.Capturer // nil records build information only
|
||||
}
|
||||
|
||||
// Create captures and stores a snapshot.
|
||||
@@ -365,21 +368,23 @@ func (s *Store) Create(ctx context.Context, m minds.Mind, name, note, by string)
|
||||
}
|
||||
raw, _ := json.Marshal(a)
|
||||
counts, _ := json.Marshal(a.Counts())
|
||||
sn := Snapshot{ID: ids.New(), MindID: m.ID, Name: name, Note: note, EventSeq: a.Source.EventSeq, Checksum: a.Checksum, SizeBytes: len(raw), Counts: a.Counts(), CreatedBy: by, CreatedAt: time.Now()}
|
||||
_, err = s.DB.Exec(ctx, `INSERT INTO snapshots (id, mind_id, name, note, event_seq, checksum, size_bytes, counts, archive, created_by, created_at) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`,
|
||||
sn.ID, sn.MindID, sn.Name, sn.Note, sn.EventSeq, sn.Checksum, sn.SizeBytes, counts, raw, sn.CreatedBy, sn.CreatedAt)
|
||||
sn := Snapshot{ID: ids.New(), MindID: m.ID, Name: name, Note: note, EventSeq: a.Source.EventSeq, Checksum: a.Checksum, SizeBytes: len(raw), Counts: a.Counts(), CreatedBy: by, CreatedAt: time.Now(), Identity: s.Identity.Capture(ctx, m.ID)}
|
||||
idj, _ := json.Marshal(sn.Identity)
|
||||
_, err = s.DB.Exec(ctx, `INSERT INTO snapshots (id, mind_id, name, note, event_seq, checksum, size_bytes, counts, archive, created_by, created_at, identity) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12)`,
|
||||
sn.ID, sn.MindID, sn.Name, sn.Note, sn.EventSeq, sn.Checksum, sn.SizeBytes, counts, raw, sn.CreatedBy, sn.CreatedAt, idj)
|
||||
return sn, err
|
||||
}
|
||||
|
||||
const snapCols = `id, mind_id, name, note, event_seq, checksum, size_bytes, counts, created_by, created_at`
|
||||
const snapCols = `id, mind_id, name, note, event_seq, checksum, size_bytes, counts, created_by, created_at, identity`
|
||||
|
||||
func scanSnap(row pgx.Row) (Snapshot, error) {
|
||||
var sn Snapshot
|
||||
var counts []byte
|
||||
if err := row.Scan(&sn.ID, &sn.MindID, &sn.Name, &sn.Note, &sn.EventSeq, &sn.Checksum, &sn.SizeBytes, &counts, &sn.CreatedBy, &sn.CreatedAt); err != nil {
|
||||
var counts, idj []byte
|
||||
if err := row.Scan(&sn.ID, &sn.MindID, &sn.Name, &sn.Note, &sn.EventSeq, &sn.Checksum, &sn.SizeBytes, &counts, &sn.CreatedBy, &sn.CreatedAt, &idj); err != nil {
|
||||
return sn, storage.Translate(err)
|
||||
}
|
||||
json.Unmarshal(counts, &sn.Counts) //nolint:errcheck
|
||||
json.Unmarshal(idj, &sn.Identity) //nolint:errcheck
|
||||
return sn, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -220,3 +220,10 @@ func withLock(ctx context.Context, pool *pgxpool.Pool, fn func(tx pgx.Tx) error)
|
||||
return fn(tx)
|
||||
})
|
||||
}
|
||||
|
||||
// SchemaVersion returns the highest applied migration version (0 when none).
|
||||
func SchemaVersion(ctx context.Context, pool *pgxpool.Pool) (int, error) {
|
||||
var v int
|
||||
err := pool.QueryRow(ctx, `SELECT coalesce(max(version), 0) FROM schema_migrations`).Scan(&v)
|
||||
return v, err
|
||||
}
|
||||
|
||||
@@ -1,5 +1,45 @@
|
||||
// Package version exposes build information injected at link time.
|
||||
// Package version exposes build information injected at link time and
|
||||
// recorded by the Go toolchain, so replays and experiments can say which
|
||||
// code produced them.
|
||||
package version
|
||||
|
||||
import (
|
||||
"runtime"
|
||||
"runtime/debug"
|
||||
)
|
||||
|
||||
// Version is set via -ldflags "-X github.com/andydixon/noema/internal/version.Version=...".
|
||||
var Version = "dev"
|
||||
|
||||
// Build describes the running binary.
|
||||
type Build struct {
|
||||
Version string `json:"version"`
|
||||
Commit string `json:"commit,omitempty"` // VCS revision when built from a checkout
|
||||
Modified bool `json:"modified,omitempty"` // working tree was dirty at build time
|
||||
BuildTime string `json:"build_time,omitempty"` // VCS commit time (reproducible; not the wall clock of the build)
|
||||
GoVersion string `json:"go_version"`
|
||||
OS string `json:"os"`
|
||||
Arch string `json:"arch"`
|
||||
}
|
||||
|
||||
// Info reads the build record. Missing VCS data (a build from a source
|
||||
// archive rather than a checkout) is left blank, never invented.
|
||||
func Info() Build {
|
||||
b := Build{Version: Version, GoVersion: runtime.Version(), OS: runtime.GOOS, Arch: runtime.GOARCH}
|
||||
if bi, ok := debug.ReadBuildInfo(); ok {
|
||||
if b.Version == "dev" && bi.Main.Version != "" && bi.Main.Version != "(devel)" {
|
||||
b.Version = bi.Main.Version
|
||||
}
|
||||
for _, s := range bi.Settings {
|
||||
switch s.Key {
|
||||
case "vcs.revision":
|
||||
b.Commit = s.Value
|
||||
case "vcs.modified":
|
||||
b.Modified = s.Value == "true"
|
||||
case "vcs.time":
|
||||
b.BuildTime = s.Value
|
||||
}
|
||||
}
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user