From 16cac754fb2c85cf9a5d1ba9fc5e9e6b7d720cae Mon Sep 17 00:00:00 2001 From: Fodesu Date: Sun, 18 Oct 2026 16:51:26 +0800 Subject: [PATCH 1/2] refactor(projection): rename commit gate capability --- agentcore/chatlog/projection.go | 10 +++--- agentcore/compaction/compaction.go | 10 +++--- agentcore/module/projection.go | 25 +++++++------- agentcore/module/registry.go | 14 ++++---- agentcore/module/registry_test.go | 14 ++++---- agentcore/session/projection.go | 8 ++--- agentcore/session/writer/cache_test.go | 12 +++---- agentcore/session/writer/health_test.go | 40 +++++++++++------------ agentcore/session/writer/projector.go | 6 ++-- agentcore/session/writer/writer.go | 4 +-- agentcore/session/writer/writer_test.go | 2 +- agentcore/sessionkernel/kernel.go | 2 +- agentcore/turn/sessionstore/projection.go | 2 +- agentcore/turn/sessionstore/surface.go | 2 +- 14 files changed, 76 insertions(+), 75 deletions(-) diff --git a/agentcore/chatlog/projection.go b/agentcore/chatlog/projection.go index d4b55c8a..d044bd92 100644 --- a/agentcore/chatlog/projection.go +++ b/agentcore/chatlog/projection.go @@ -120,8 +120,8 @@ var SurfaceProjection = module.ProjectionDefinition{ Consumes: chatlogConsumes, // Assistant and tool_result entries are projected from Turn facts // a fork inherits that conversation content. - Inherits: module.InheritAll, - Authoritative: true, + Inherits: module.InheritAll, + RequiredForCommit: true, Initial: func() (any, error) { return Surface{}, nil }, @@ -402,9 +402,9 @@ type Context struct { var ContextProjection = module.ProjectionDefinition{ ID: ContextProjectionID, Version: 1, - Consumes: chatlogConsumes, - Inherits: module.InheritAll, - Authoritative: true, + Consumes: chatlogConsumes, + Inherits: module.InheritAll, + RequiredForCommit: true, Initial: func() (any, error) { return Context{Pending: map[InputID]Input{}, Superseded: map[ToolResultID]ToolResultID{}}, nil }, diff --git a/agentcore/compaction/compaction.go b/agentcore/compaction/compaction.go index 8661259e..6870c8b5 100644 --- a/agentcore/compaction/compaction.go +++ b/agentcore/compaction/compaction.go @@ -152,11 +152,11 @@ var requestBindings = module.BindingReferenceDefinition{ // its parent's in-flight summary execution or acknowledgement responsibility. var Projection = module.ProjectionDefinition{ ID: ProjectionID, Version: 1, - Consumes: []ledger.EventType{TypeRequested, TypeFinished, TypeAcknowledged}, - Authoritative: true, - Initial: func() (any, error) { return State{Operations: map[ID]Operation{}}, nil }, - StateCodec: module.JSONStateCodec[State]{}, - Apply: apply, + Consumes: []ledger.EventType{TypeRequested, TypeFinished, TypeAcknowledged}, + RequiredForCommit: true, + Initial: func() (any, error) { return State{Operations: map[ID]Operation{}}, nil }, + StateCodec: module.JSONStateCodec[State]{}, + Apply: apply, } var Module = module.ModuleDescriptor{ diff --git a/agentcore/module/projection.go b/agentcore/module/projection.go index 61f9fee6..751be5fb 100644 --- a/agentcore/module/projection.go +++ b/agentcore/module/projection.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "github.com/felinics/twilight/agentcore/ledger" "github.com/felinics/twilight/agentcore/jsonstable" @@ -26,14 +27,14 @@ type ProjectionDefinition struct { // facts declares InheritAll, or names the domains it takes with // InheritStreams. Inherits InheritPolicy - // Authoritative marks a projection a module guards its stream's - // invariants with: a provisional commit its fold cannot fold is refused - // before anything is persisted. Every other projection is a derived read - // model: a fold failure marks it unhealthy for the Writer's lifetime and - // never blocks the facts. Write-time invariants belong to - // the Writer's decision, not to projections; an authoritative fold + // RequiredForCommit marks a projection whose fold must accept a + // provisional commit before anything is persisted. Replay failures also + // prevent the Writer from opening. Only trusted core modules may set it. + // Every other projection is a derived read model: a fold failure marks it + // Writer's lifetime and never blocks the facts. Write-time invariants + // belong to the Writer's decision, not to projections; a required fold // failing is a defect, not a business rejection. - Authoritative bool + RequiredForCommit bool } // InheritPolicy decides whether a projection folds the batches of one @@ -167,29 +168,29 @@ func (r *Registry) applyEvent(s *ProjectionScope, state any, pos ledger.Position return next, nil } -// checkpointDomain namespaces the digest of an authoritative checkpoint. +// checkpointDomain namespaces the digest of a required-for-commit checkpoint. const checkpointDomain = "twilight/projection-checkpoint" -// sealedCheckpoint is the cache Value of an authoritative projection: the +// sealedCheckpoint is the cache Value of a required-for-commit projection: the // encoded state and a digest over (projection, version, through, state). type sealedCheckpoint struct { State json.RawMessage `json:"state"` Digest jsonstable.Digest `json:"digest"` } -// checkpointDigest is the digest an authoritative entry must carry. +// checkpointDigest is the digest a required-for-commit entry must carry. func checkpointDigest(id ProjectionID, v ProjectionVersion, through ledger.Head, state jsonstable.Value) jsonstable.Digest { preimage, _ := jsonstable.EncodeTypedPayload(1, checkpointDomain, []string{string(id), fmt.Sprint(uint64(v)), fmt.Sprint(uint64(through.Next)), string(state.Bytes())}) return jsonstable.DigestBytes(preimage) } -// SealCheckpoint wraps an authoritative projection's encoded state with its +// SealCheckpoint wraps a required-for-commit projection's encoded state with its // digest for the cache. func SealCheckpoint(id ProjectionID, v ProjectionVersion, through ledger.Head, state jsonstable.Value) (jsonstable.Value, error) { return jsonstable.FromValue(sealedCheckpoint{State: state.Bytes(), Digest: checkpointDigest(id, v, through, state)}) } -// OpenCheckpoint unwraps a sealed authoritative entry, verifying its digest; +// OpenCheckpoint unwraps a sealed required-for-commit entry, verifying its digest; // ok is false when the entry is not a checkpoint of this projection at // through, and the caller treats it as absent. func OpenCheckpoint(id ProjectionID, v ProjectionVersion, through ledger.Head, sealed jsonstable.Value) (jsonstable.Value, bool) { diff --git a/agentcore/module/registry.go b/agentcore/module/registry.go index 967602bc..ca928b48 100644 --- a/agentcore/module/registry.go +++ b/agentcore/module/registry.go @@ -182,7 +182,7 @@ func validSegment(kind, v string) error { // BuildRegistry validates the module set and freezes the indexes. Every // module passed here is trusted: the caller vouches for it, so it may -// declare authoritative projections. Modules from outside the +// declare required-for-commit projections. Modules from outside the // deployment's trust boundary go through BuildRegistryWithExtensions. func BuildRegistry(modules ...ModuleDescriptor) (*Registry, error) { return BuildRegistryWithExtensions(modules, nil) @@ -190,8 +190,8 @@ func BuildRegistry(modules ...ModuleDescriptor) (*Registry, error) { // BuildRegistryWithExtensions builds a registry from trusted core modules // and untrusted extensions. Trust is a property of how a module reached the -// registry, not of what its descriptor says: an extension may not declare an -// authoritative projection and may not use the SourceTwilight source, so no +// registry, not of what its descriptor says: an extension may not declare a +// required-for-commit projection and may not use the SourceTwilight source, so no // descriptor can claim the first-party capability for itself. func BuildRegistryWithExtensions(core, extensions []ModuleDescriptor) (*Registry, error) { modules := make([]ModuleDescriptor, 0, len(core)+len(extensions)) @@ -206,8 +206,8 @@ func BuildRegistryWithExtensions(core, extensions []ModuleDescriptor) (*Registry return nil, &ledger.Error{Code: ledger.CodeInvalid, Detail: fmt.Sprintf("extension module %q claims the %s source", m.ID, SourceTwilight)} } for _, p := range m.Projections { - if p.Authoritative { - return nil, &ledger.Error{Code: ledger.CodeInvalid, Detail: fmt.Sprintf("projection %q of extension %s/%s declares Authoritative; only trusted core modules may", p.ID, m.Source, m.ID)} + if p.RequiredForCommit { + return nil, &ledger.Error{Code: ledger.CodeInvalid, Detail: fmt.Sprintf("projection %q of extension %s/%s declares RequiredForCommit; only trusted core modules may", p.ID, m.Source, m.ID)} } } modules = append(modules, *m) @@ -253,8 +253,8 @@ func BuildRegistryWithExtensions(core, extensions []ModuleDescriptor) (*Registry // Refusing commits is a capability of trusted core modules, not // something a descriptor declares for itself; the // extension path above already refused it, this guards the map. - if p.Authoritative && !trusted[key] { - return nil, &ledger.Error{Code: ledger.CodeInvalid, Detail: fmt.Sprintf("projection %q of module %s/%s declares Authoritative without trust", p.ID, m.Source, m.ID)} + if p.RequiredForCommit && !trusted[key] { + return nil, &ledger.Error{Code: ledger.CodeInvalid, Detail: fmt.Sprintf("projection %q of module %s/%s declares RequiredForCommit without trust", p.ID, m.Source, m.ID)} } k := projectionKey{p.ID, p.Version} if _, dup := r.projections[k]; dup { diff --git a/agentcore/module/registry_test.go b/agentcore/module/registry_test.go index 10f465e2..22f50882 100644 --- a/agentcore/module/registry_test.go +++ b/agentcore/module/registry_test.go @@ -275,25 +275,25 @@ func TestRegistryMultiVersionCodecsCoexist(t *testing.T) { } } -// Authoritative is a capability of trusted core modules. An +// RequiredForCommit is a capability of trusted core modules. An // extension cannot declare it, nor claim the first-party source to get it. -func TestExtensionsCannotBeAuthoritative(t *testing.T) { +func TestExtensionsCannotDeclareRequiredForCommit(t *testing.T) { core := ModuleDescriptor{Source: SourceTwilight, ID: "core", Projections: []ProjectionDefinition{{ - ID: "twilight/core/p", Version: 1, Authoritative: true, + ID: "twilight/core/p", Version: 1, RequiredForCommit: true, Initial: func() (any, error) { return struct{}{}, nil }, Apply: func(s any, _ DecodedEvent) (any, error) { return s, nil }, StateCodec: JSONStateCodec[struct{}]{}}}} ext := ModuleDescriptor{Source: "acme", ID: "plugin", Projections: []ProjectionDefinition{{ - ID: "acme/plugin/p", Version: 1, Authoritative: true, + ID: "acme/plugin/p", Version: 1, RequiredForCommit: true, Initial: func() (any, error) { return struct{}{}, nil }, Apply: func(s any, _ DecodedEvent) (any, error) { return s, nil }, StateCodec: JSONStateCodec[struct{}]{}}}} if _, err := BuildRegistryWithExtensions([]ModuleDescriptor{core}, []ModuleDescriptor{ext}); err == nil { - t.Fatal("extension declared an authoritative projection") + t.Fatal("extension declared a required-for-commit projection") } impostor := ext impostor.Source = SourceTwilight - impostor.Projections[0].Authoritative = false + impostor.Projections[0].RequiredForCommit = false if _, err := BuildRegistryWithExtensions([]ModuleDescriptor{core}, []ModuleDescriptor{impostor}); err == nil { t.Fatal("extension claimed the twilight source") } - ext.Projections[0].Authoritative = false + ext.Projections[0].RequiredForCommit = false if _, err := BuildRegistryWithExtensions([]ModuleDescriptor{core}, []ModuleDescriptor{ext}); err != nil { t.Fatalf("derived extension refused: %v", err) } diff --git a/agentcore/session/projection.go b/agentcore/session/projection.go index 76f4c436..a2419e3f 100644 --- a/agentcore/session/projection.go +++ b/agentcore/session/projection.go @@ -20,7 +20,7 @@ type ProjectionReader interface { // ProjectionCache is the optional derived cache. Entries may be // lost or stale at any time; readers verify Through against the stream. -// An Authoritative projection's entry is a checkpoint the Writer plans the +// A required-for-commit projection's entry is a checkpoint the Writer plans the // next commit: its state bytes travel with the digest // the Writer computed when it saved them, in the same Value (see // module.SealCheckpoint), and a reader that finds the digest missing or @@ -101,7 +101,7 @@ func SaveProjection(ctx context.Context, cache ProjectionCache, registry *module if err != nil { return err } - if def.Authoritative { + if def.RequiredForCommit { if encoded, err = module.SealCheckpoint(id, v, through, encoded); err != nil { return err } @@ -110,14 +110,14 @@ func SaveProjection(ctx context.Context, cache ProjectionCache, registry *module } // LoadProjectionEntry reads a cache entry for def, unsealing and verifying -// an authoritative one; an entry that does not verify is a +// a required-for-commit one; an entry that does not verify is a // miss. func LoadProjectionEntry(ctx context.Context, cache ProjectionCache, def *module.ProjectionDefinition, sid SessionID) (jsonstable.Value, ledger.Head, bool, error) { encoded, through, ok, err := cache.Load(ctx, sid, def.ID, def.Version) if err != nil || !ok { return jsonstable.Value{}, ledger.Head{}, false, err } - if def.Authoritative { + if def.RequiredForCommit { state, verified := module.OpenCheckpoint(def.ID, def.Version, through, encoded) if !verified { return jsonstable.Value{}, ledger.Head{}, false, nil diff --git a/agentcore/session/writer/cache_test.go b/agentcore/session/writer/cache_test.go index 740146d4..146f76c7 100644 --- a/agentcore/session/writer/cache_test.go +++ b/agentcore/session/writer/cache_test.go @@ -391,18 +391,18 @@ func TestWriterWithoutCacheFoldsEverything(t *testing.T) { } } -// an authoritative projection's entry is a checkpoint sealed +// a required-for-commit projection's entry is a checkpoint sealed // with a digest over its identity, position and state. A well-formed entry // whose state was not what the Writer saved (here: a derived projection's -// entry copied under the authoritative one) is a miss, and the Writer +// entry copied under the required-for-commit one) is a miss, and the Writer // refolds from the log; a derived projection's entry stays a plain state. -func TestAuthoritativeEntryMustVerify(t *testing.T) { +func TestRequiredForCommitEntryMustVerify(t *testing.T) { ctx := context.Background() counter := newApplyCounter() typ := tpfx("k") + "row" - mk := func(id module.ProjectionID, authoritative bool) module.ProjectionDefinition { + mk := func(id module.ProjectionID, requiredForCommit bool) module.ProjectionDefinition { return module.ProjectionDefinition{ - ID: id, Version: 1, Consumes: []ledger.EventType{typ}, Authoritative: authoritative, + ID: id, Version: 1, Consumes: []ledger.EventType{typ}, RequiredForCommit: requiredForCommit, Initial: func() (any, error) { return noteState{}, nil }, Apply: func(state any, e module.DecodedEvent) (any, error) { counter.inc(id) @@ -429,7 +429,7 @@ func TestAuthoritativeEntryMustVerify(t *testing.T) { if err := w.Close(ctx); err != nil { t.Fatal(err) } - // Both entries are at head; the authoritative one is sealed. + // Both entries are at head; the required-for-commit one is sealed. sealed, through, ok, err := f.cache.Load(ctx, "s", alphaID, 1) if err != nil || !ok || through.Next != 2 { t.Fatalf("alpha entry = ok:%v through:%d %v", ok, through.Next, err) diff --git a/agentcore/session/writer/health_test.go b/agentcore/session/writer/health_test.go index f37382ed..4de943af 100644 --- a/agentcore/session/writer/health_test.go +++ b/agentcore/session/writer/health_test.go @@ -11,13 +11,13 @@ import ( "testing" ) -// healthModule has one event type and two projections over it: an -// authoritative one and a derived one, both of which fail on the text "boom". +// healthModule has one event type and two projections over it: a +// required-for-commit one and a derived one, both of which fail on the text "boom". func healthModule() module.ModuleDescriptor { typ := tpfx("h") + "row" - mk := func(id module.ProjectionID, authoritative bool) module.ProjectionDefinition { + mk := func(id module.ProjectionID, requiredForCommit bool) module.ProjectionDefinition { return module.ProjectionDefinition{ - ID: id, Version: 1, Consumes: []ledger.EventType{typ}, Authoritative: authoritative, + ID: id, Version: 1, Consumes: []ledger.EventType{typ}, RequiredForCommit: requiredForCommit, Initial: func() (any, error) { return noteState{}, nil }, Apply: func(state any, e module.DecodedEvent) (any, error) { text := e.Value.(notePayload).Text @@ -33,12 +33,12 @@ func healthModule() module.ModuleDescriptor { } return module.ModuleDescriptor{Source: module.SourceTwilight, ID: "h", Streams: noteStreams(), Events: []module.EventDefinition{{Type: typ, Domain: noteDomain, Codecs: map[module.PayloadVersion]module.PayloadCodec{1: module.JSONCodec[notePayload]{}}}}, - Projections: []module.ProjectionDefinition{mk("h/authoritative", true), mk("h/derived", false)}} + Projections: []module.ProjectionDefinition{mk("h/required", true), mk("h/derived", false)}} } // a derived projection that cannot fold a commit does not refuse // the commit; it is marked unhealthy, reads report it, and it is not cached. -// An authoritative projection that cannot fold refuses the commit. +// A required-for-commit projection that cannot fold refuses the commit. func TestDerivedProjectionFailureDoesNotBlockCommit(t *testing.T) { ctx := context.Background() store := filestoretest.Store(t) @@ -62,12 +62,12 @@ func TestDerivedProjectionFailureDoesNotBlockCommit(t *testing.T) { if res, err := commit("c1", "one", ""); err != nil || res.Outcome != CommitApplied { t.Fatalf("c1 = %+v %v", res, err) } - // Both projections fail on boom, so the authoritative one refuses. + // Both projections fail on boom, so the required-for-commit one refuses. if res, err := commit("c2", "boom", ""); err != nil || res.Outcome != CommitInvalid { - t.Fatalf("authoritative refusal = %+v %v", res, err) + t.Fatalf("required-for-commit refusal = %+v %v", res, err) } // Reopen with only the derived projection failing: rebuild the registry - // with an authoritative projection that accepts everything. + // with a required-for-commit projection that accepts everything. registry2, err := module.BuildRegistry(func() module.ModuleDescriptor { m := healthModule() m.Projections[0].Apply = func(state any, e module.DecodedEvent) (any, error) { return state, nil } @@ -89,8 +89,8 @@ func TestDerivedProjectionFailureDoesNotBlockCommit(t *testing.T) { if _, _, err := w.Projections().Load(ctx, "s", "h/derived", 1); !errors.Is(err, &ledger.Error{Code: ledger.CodeProjectionUnhealthy}) { t.Fatalf("derived read = %v, want unhealthy", err) } - if _, _, err := w.Projections().Load(ctx, "s", "h/authoritative", 1); err != nil { - t.Fatalf("authoritative read after a derived failure: %v", err) + if _, _, err := w.Projections().Load(ctx, "s", "h/required", 1); err != nil { + t.Fatalf("required-for-commit read after a derived failure: %v", err) } // The next commit still lands and the unhealthy projection stays behind. if res, err := commit("c3", "three", ""); err != nil || res.Outcome != CommitApplied { @@ -99,14 +99,14 @@ func TestDerivedProjectionFailureDoesNotBlockCommit(t *testing.T) { if _, through, ok, _ := cache.Load(ctx, "s", "h/derived", 1); ok && through.Next > 1 { t.Fatalf("unhealthy projection cached at head %+v", through) } - if _, through, ok, _ := cache.Load(ctx, "s", "h/authoritative", 1); !ok || through.Next != 3 { + if _, through, ok, _ := cache.Load(ctx, "s", "h/required", 1); !ok || through.Next != 3 { t.Fatalf("healthy projection cache = ok=%v %+v, want head 3", ok, through) } } // across a reopen: a derived projection that cannot fold the log // does not keep the Session from opening; it stops at its last good commit -// and is unhealthy, the authoritative projections fold to head. +// and is unhealthy, the required-for-commit projections fold to head. func TestDerivedProjectionFailureDoesNotBlockReopen(t *testing.T) { ctx := context.Background() store := filestoretest.Store(t) @@ -138,7 +138,7 @@ func TestDerivedProjectionFailureDoesNotBlockReopen(t *testing.T) { t.Fatal(err) } strict := healthModule() - strict.Projections[0].Apply = permissive.Projections[0].Apply // authoritative one keeps folding + strict.Projections[0].Apply = permissive.Projections[0].Apply // required-for-commit one keeps folding reg2, err := module.BuildRegistry(strict) if err != nil { t.Fatal(err) @@ -150,8 +150,8 @@ func TestDerivedProjectionFailureDoesNotBlockReopen(t *testing.T) { if _, _, err := w.Projections().Load(ctx, "s", "h/derived", 1); !errors.Is(err, &ledger.Error{Code: ledger.CodeProjectionUnhealthy}) { t.Fatalf("derived after reopen = %v, want unhealthy", err) } - if _, head, err := w.Projections().Load(ctx, "s", "h/authoritative", 1); err != nil || head.Next != 3 { - t.Fatalf("authoritative after reopen = %+v %v", head, err) + if _, head, err := w.Projections().Load(ctx, "s", "h/required", 1); err != nil || head.Next != 3 { + t.Fatalf("required-for-commit after reopen = %+v %v", head, err) } // And the Session still writes. if res, err := w.Commit(ctx, func(View) (*SemanticGroup, error) { @@ -159,14 +159,14 @@ func TestDerivedProjectionFailureDoesNotBlockReopen(t *testing.T) { }); err != nil || res.Outcome != CommitApplied { t.Fatalf("commit after reopen = %+v %v", res, err) } - // An authoritative projection that cannot fold the log refuses the open. - failingAuth := healthModule() - reg3, err := module.BuildRegistry(failingAuth) + // A required-for-commit projection that cannot fold the log refuses the open. + failingRequired := healthModule() + reg3, err := module.BuildRegistry(failingRequired) if err != nil { t.Fatal(err) } _ = w.Close(ctx) if _, err := acquireTestWriterConfig(ctx, store, reg3, Admission{}, "s", session.OpenOptions{Takeover: true}, WritersConfig{}); err == nil { - t.Fatal("reopen with a failing authoritative projection succeeded") + t.Fatal("reopen with a failing required-for-commit projection succeeded") } } diff --git a/agentcore/session/writer/projector.go b/agentcore/session/writer/projector.go index e1d470de..080aa0ec 100644 --- a/agentcore/session/writer/projector.go +++ b/agentcore/session/writer/projector.go @@ -97,7 +97,7 @@ func (p *projector) resume(page *session.CommitPage, from ledger.CommitSeq) erro } folded, err := p.registry.FoldFrom(scope, p.states[k], page.Commits[off:], page.Header.Seed()) if err != nil { - if scope.Def.Authoritative { + if scope.Def.RequiredForCommit { return err } // A derived projection that cannot fold the log does not keep @@ -165,7 +165,7 @@ type folded struct { } // fold pre-folds one provisional commit into every projection without -// installing anything. An authoritative projection that refuses the commit +// installing anything. A required-for-commit projection that refuses the commit // refuses it for the Writer; a derived one keeps its last good state and is // marked unhealthy once the commit lands. A projection already // unhealthy is not folded further. @@ -177,7 +177,7 @@ func (p *projector) fold(provisional ledger.Commit) (folded, error) { } state, err := p.registry.Fold(scope, p.states[k], []ledger.Commit{provisional}) if err != nil { - if scope.Def.Authoritative { + if scope.Def.RequiredForCommit { return folded{}, err } if out.failed == nil { diff --git a/agentcore/session/writer/writer.go b/agentcore/session/writer/writer.go index bb438acf..7522da75 100644 --- a/agentcore/session/writer/writer.go +++ b/agentcore/session/writer/writer.go @@ -298,11 +298,11 @@ func (w *sessionWriter) Commit(ctx context.Context, fn CommitFn) (CommitResult, // same operation and is not written again. return CommitResult{Outcome: CommitAlreadyApplied, Commit: existing}, nil } - // Projections must accept the commit before anything is persisted. The + // Required projections must accept the commit before anything is persisted. The // provisional commit is what a reader folds: the kernel assigns only Seq // inside Append. provisional := ledger.Proposal{CommitID: group.CommitID, Batches: batches}.At(w.kernel.Head().Next) - // Only an authoritative projection's fold refuses the commit; a derived + // Only a required-for-commit projection's fold refuses the commit; a derived // one that cannot fold is marked unhealthy once the commit lands. next, err := w.projections.fold(provisional) if err != nil { diff --git a/agentcore/session/writer/writer_test.go b/agentcore/session/writer/writer_test.go index b43d5b80..b49ba963 100644 --- a/agentcore/session/writer/writer_test.go +++ b/agentcore/session/writer/writer_test.go @@ -55,7 +55,7 @@ func noteModule(id module.ModuleID, requires ...module.ModuleRequirement) module {Type: tpfx(id) + "hint", Domain: noteDomain, Codecs: map[module.PayloadVersion]module.PayloadCodec{1: module.JSONCodec[notePayload]{}}, Ignorable: true}, }, Projections: []module.ProjectionDefinition{{ - ID: module.ProjectionID(string(typ) + "s"), Version: 1, Consumes: []ledger.EventType{typ}, Authoritative: true, + ID: module.ProjectionID(string(typ) + "s"), Version: 1, Consumes: []ledger.EventType{typ}, RequiredForCommit: true, Initial: func() (any, error) { return noteState{}, nil }, Apply: func(state any, e module.DecodedEvent) (any, error) { s := state.(noteState) diff --git a/agentcore/sessionkernel/kernel.go b/agentcore/sessionkernel/kernel.go index cfe853e2..0143c843 100644 --- a/agentcore/sessionkernel/kernel.go +++ b/agentcore/sessionkernel/kernel.go @@ -172,7 +172,7 @@ func New(p Ports) (*Kernel, error) { //nolint:gocritic // hugeParam: Ports is a // NewRegistry builds the module registry of a Kernel: the first-party // modules as trusted core, extensions after them. Extensions cannot declare -// authoritative projections. +// required-for-commit projections. func NewRegistry(extensions []module.ModuleDescriptor) (*module.Registry, error) { return module.BuildRegistryWithExtensions( []module.ModuleDescriptor{chatlog.Module, sessionstore.Module, compaction.Module}, extensions) diff --git a/agentcore/turn/sessionstore/projection.go b/agentcore/turn/sessionstore/projection.go index 4be157ed..178df2e2 100644 --- a/agentcore/turn/sessionstore/projection.go +++ b/agentcore/turn/sessionstore/projection.go @@ -152,7 +152,7 @@ func (machineCodec) Decode(wr jsonstable.Value) (any, error) { // MachineProjection consumes every twilight/turn/ event. var MachineProjection = module.ProjectionDefinition{ ID: MachineProjectionID, Version: 1, - Consumes: AllTypes(), Authoritative: true, + Consumes: AllTypes(), RequiredForCommit: true, Inherits: func(ledger.Domain) bool { return false }, Initial: func() (any, error) { return newMachine(), nil }, Apply: func(state any, e module.DecodedEvent) (any, error) { diff --git a/agentcore/turn/sessionstore/surface.go b/agentcore/turn/sessionstore/surface.go index 72536ee9..1e47fd87 100644 --- a/agentcore/turn/sessionstore/surface.go +++ b/agentcore/turn/sessionstore/surface.go @@ -53,7 +53,7 @@ func (s *TurnSurface) Active() (TurnView, bool) { var SurfaceProjection = module.ProjectionDefinition{ ID: SurfaceProjectionID, Version: 1, Consumes: []ledger.EventType{TypeStarted, Prefix + "input_accepted", Prefix + "turn_ended"}, - Inherits: module.InheritAll, Authoritative: true, + Inherits: module.InheritAll, RequiredForCommit: true, Initial: func() (any, error) { return TurnSurface{Turns: map[turn.TurnID]TurnView{}}, nil }, StateCodec: module.JSONStateCodec[TurnSurface]{}, Apply: applySurface, From d0d62ef9786f517f62478e6b3705a8d4160bf803 Mon Sep 17 00:00:00 2001 From: Fodesu Date: Sun, 18 Oct 2026 17:21:37 +0800 Subject: [PATCH 2/2] refactor(projection): make inheritance overrides explicit --- agentcore/chatlog/projection.go | 4 +- agentcore/compaction/compaction.go | 11 +-- agentcore/module/projection.go | 43 ++++------- agentcore/module/projection_test.go | 88 +++++++++++++++++++++++ agentcore/turn/sessionstore/projection.go | 4 +- agentcore/turn/sessionstore/surface.go | 10 +-- 6 files changed, 117 insertions(+), 43 deletions(-) create mode 100644 agentcore/module/projection_test.go diff --git a/agentcore/chatlog/projection.go b/agentcore/chatlog/projection.go index d044bd92..06800bb6 100644 --- a/agentcore/chatlog/projection.go +++ b/agentcore/chatlog/projection.go @@ -119,8 +119,7 @@ var SurfaceProjection = module.ProjectionDefinition{ ID: SurfaceProjectionID, Version: 1, Consumes: chatlogConsumes, // Assistant and tool_result entries are projected from Turn facts - // a fork inherits that conversation content. - Inherits: module.InheritAll, + // whose domain defaults preserve conversation content across a fork. RequiredForCommit: true, Initial: func() (any, error) { return Surface{}, nil @@ -403,7 +402,6 @@ type Context struct { var ContextProjection = module.ProjectionDefinition{ ID: ContextProjectionID, Version: 1, Consumes: chatlogConsumes, - Inherits: module.InheritAll, RequiredForCommit: true, Initial: func() (any, error) { return Context{Pending: map[InputID]Input{}, Superseded: map[ToolResultID]ToolResultID{}}, nil diff --git a/agentcore/compaction/compaction.go b/agentcore/compaction/compaction.go index 6870c8b5..7e980778 100644 --- a/agentcore/compaction/compaction.go +++ b/agentcore/compaction/compaction.go @@ -152,11 +152,12 @@ var requestBindings = module.BindingReferenceDefinition{ // its parent's in-flight summary execution or acknowledgement responsibility. var Projection = module.ProjectionDefinition{ ID: ProjectionID, Version: 1, - Consumes: []ledger.EventType{TypeRequested, TypeFinished, TypeAcknowledged}, - RequiredForCommit: true, - Initial: func() (any, error) { return State{Operations: map[ID]Operation{}}, nil }, - StateCodec: module.JSONStateCodec[State]{}, - Apply: apply, + Consumes: []ledger.EventType{TypeRequested, TypeFinished, TypeAcknowledged}, + RequiredForCommit: true, + InheritanceOverride: module.InheritNone, + Initial: func() (any, error) { return State{Operations: map[ID]Operation{}}, nil }, + StateCodec: module.JSONStateCodec[State]{}, + Apply: apply, } var Module = module.ModuleDescriptor{ diff --git a/agentcore/module/projection.go b/agentcore/module/projection.go index 751be5fb..2dbe911b 100644 --- a/agentcore/module/projection.go +++ b/agentcore/module/projection.go @@ -18,15 +18,11 @@ type ProjectionDefinition struct { Initial func() (any, error) Apply func(any, DecodedEvent) (any, error) StateCodec PayloadCodec - // Inherits decides, per logical stream, what the fold takes from the - // commits a fork inherits. nil follows the lineage each - // stream's domain declared: inherited batches of Inherited domains - // are folded and those of Own domains are skipped, so a - // child never interprets its parent's execution history as its own. A - // projection whose content lives in another module's segment-lineage - // facts declares InheritAll, or names the domains it takes with - // InheritStreams. - Inherits InheritPolicy + // InheritanceOverride decides which batches from an inherited prefix this + // projection folds. A nil override follows the owning stream domain's + // Inheritance declaration. InheritAll and InheritNone override it for every + // domain. + InheritanceOverride InheritancePolicy // RequiredForCommit marks a projection whose fold must accept a // provisional commit before anything is persisted. Replay failures also // prevent the Writer from opening. Only trusted core modules may set it. @@ -37,31 +33,22 @@ type ProjectionDefinition struct { RequiredForCommit bool } -// InheritPolicy decides whether a projection folds the batches of one -// logical stream from a fork's inherited prefix. -type InheritPolicy func(ledger.Domain) bool +// InheritancePolicy selects how a projection reads an inherited prefix for a +// particular stream domain. A nil policy means the domain's declaration is +// used instead. +type InheritancePolicy func(ledger.Domain) bool // InheritAll folds every batch of inherited commits. func InheritAll(ledger.Domain) bool { return true } -// InheritStreams folds the listed stream domains of inherited commits. -func InheritStreams(domains ...string) InheritPolicy { - return func(stream ledger.Domain) bool { - for _, d := range domains { - if stream.Name == d { - return true - } - } - return false - } -} +// InheritNone skips every batch of inherited commits. +func InheritNone(ledger.Domain) bool { return false } -// inherits applies the definition's policy; nil follows the lineage the -// stream's domain declared, and a domain no module declared is not -// inherited. +// inherits applies the projection policy. A missing policy follows the +// inheritance declared by the stream's owning module. func (r *Registry) inherits(d *ProjectionDefinition, stream ledger.Domain) bool { - if d.Inherits != nil { - return d.Inherits(stream) + if d.InheritanceOverride != nil { + return d.InheritanceOverride(stream) } _, def, ok := r.LookupStream(stream.Name) return ok && def.Inheritance == Inherited diff --git a/agentcore/module/projection_test.go b/agentcore/module/projection_test.go new file mode 100644 index 00000000..21cc83fc --- /dev/null +++ b/agentcore/module/projection_test.go @@ -0,0 +1,88 @@ +package module + +import ( + "testing" + + "github.com/felinics/twilight/agentcore/ledger" +) + +func TestProjectionInheritanceOverride(t *testing.T) { + const ( + inheritedEvent ledger.EventType = "twilight/history/inherited" + ownEvent ledger.EventType = "twilight/history/own" + ) + r, err := BuildRegistry(ModuleDescriptor{ + Source: SourceTwilight, + ID: "history", + Streams: []StreamDefinition{ + {Domain: "inherited", Inheritance: Inherited}, + {Domain: "own", Inheritance: Own}, + }, + Events: []EventDefinition{ + {Type: inheritedEvent, Domain: "inherited", Codecs: map[PayloadVersion]PayloadCodec{1: JSONCodec[struct{}]{}}}, + {Type: ownEvent, Domain: "own", Codecs: map[PayloadVersion]PayloadCodec{1: JSONCodec[struct{}]{}}}, + }, + Projections: []ProjectionDefinition{{ + ID: "twilight/history/projection", + Version: 1, + Consumes: []ledger.EventType{inheritedEvent, ownEvent}, + Initial: func() (any, error) { return []string{}, nil }, + Apply: func(state any, event DecodedEvent) (any, error) { + return append(state.([]string), event.Domain.Name), nil + }, + StateCodec: JSONStateCodec[[]string]{}, + }}, + }) + if err != nil { + t.Fatal(err) + } + scope, err := r.ScopeFor("twilight/history/projection", 1) + if err != nil { + t.Fatal(err) + } + + inheritedPayload, err := r.Encode(inheritedEvent, struct{}{}) + if err != nil { + t.Fatal(err) + } + inherited := ledger.Event{Type: inheritedEvent, Payload: inheritedPayload} + ownPayload, err := r.Encode(ownEvent, struct{}{}) + if err != nil { + t.Fatal(err) + } + own := ledger.Event{Type: ownEvent, Payload: ownPayload} + commits := []ledger.Commit{{ + Seq: 0, + Batches: []ledger.EventBatch{ + {Domain: ledger.Domain{Name: "inherited"}, Events: []ledger.Event{inherited}}, + {Domain: ledger.Domain{Name: "own"}, Events: []ledger.Event{own}}, + }, + }} + + for _, tc := range []struct { + name string + policy InheritancePolicy + expected []string + }{ + {name: "domain defaults", expected: []string{"inherited"}}, + {name: "inherit all", policy: InheritAll, expected: []string{"inherited", "own"}}, + {name: "inherit none", policy: InheritNone}, + } { + t.Run(tc.name, func(t *testing.T) { + scope.Def.InheritanceOverride = tc.policy + state, err := r.FoldFrom(scope, []string{}, commits, ledger.Head{Next: 1}) + if err != nil { + t.Fatal(err) + } + got := state.([]string) + if len(got) != len(tc.expected) { + t.Fatalf("folded domains = %v, want %v", got, tc.expected) + } + for i := range got { + if got[i] != tc.expected[i] { + t.Fatalf("folded domains = %v, want %v", got, tc.expected) + } + } + }) + } +} diff --git a/agentcore/turn/sessionstore/projection.go b/agentcore/turn/sessionstore/projection.go index 178df2e2..acbcbba7 100644 --- a/agentcore/turn/sessionstore/projection.go +++ b/agentcore/turn/sessionstore/projection.go @@ -153,8 +153,8 @@ func (machineCodec) Decode(wr jsonstable.Value) (any, error) { var MachineProjection = module.ProjectionDefinition{ ID: MachineProjectionID, Version: 1, Consumes: AllTypes(), RequiredForCommit: true, - Inherits: func(ledger.Domain) bool { return false }, - Initial: func() (any, error) { return newMachine(), nil }, + InheritanceOverride: module.InheritNone, + Initial: func() (any, error) { return newMachine(), nil }, Apply: func(state any, e module.DecodedEvent) (any, error) { m, ok := state.(Machine) if !ok { diff --git a/agentcore/turn/sessionstore/surface.go b/agentcore/turn/sessionstore/surface.go index 1e47fd87..15cdeb0d 100644 --- a/agentcore/turn/sessionstore/surface.go +++ b/agentcore/turn/sessionstore/surface.go @@ -52,11 +52,11 @@ func (s *TurnSurface) Active() (TurnView, bool) { // History inherits facts; execution's MachineProjection explicitly does not. var SurfaceProjection = module.ProjectionDefinition{ ID: SurfaceProjectionID, Version: 1, - Consumes: []ledger.EventType{TypeStarted, Prefix + "input_accepted", Prefix + "turn_ended"}, - Inherits: module.InheritAll, RequiredForCommit: true, - Initial: func() (any, error) { return TurnSurface{Turns: map[turn.TurnID]TurnView{}}, nil }, - StateCodec: module.JSONStateCodec[TurnSurface]{}, - Apply: applySurface, + Consumes: []ledger.EventType{TypeStarted, Prefix + "input_accepted", Prefix + "turn_ended"}, + RequiredForCommit: true, + Initial: func() (any, error) { return TurnSurface{Turns: map[turn.TurnID]TurnView{}}, nil }, + StateCodec: module.JSONStateCodec[TurnSurface]{}, + Apply: applySurface, } func applySurface(state any, e module.DecodedEvent) (any, error) {