Documentation
¶
Overview ¶
Package session implements the client-facing Sessions gRPC service, backed by the durable sqlitelog journal and the event-sourced controller. It is the top seam a producer (GitHub, Foundry, a custom app) or the agentctl CLI drives. For M0 it hosts a single harness (echo); the model and harness are injected.
A session UID is a key in the sqlitelog store. CreateSession persists a metadata row up front, so a session exists — and is listable — from the moment it is created, before it has any events and before any compute is provisioned. The log remains the authority for everything the log knows (cursor, lifecycle); the metadata row carries only what the log cannot express, such as the project, display name, and configured harness/model.
Index ¶
- Constants
- func ReserveStaticHarnessNames(store *sqlitelog.Store, registry *placement.Registry) error
- type HarnessRegistry
- func (h *HarnessRegistry) Close()
- func (h *HarnessRegistry) GetHarness(ctx context.Context, req *v1.GetHarnessRequest) (*v1.HarnessRegistration, error)
- func (h *HarnessRegistry) ListHarnesses(ctx context.Context, req *v1.ListHarnessesRequest) (*v1.ListHarnessesResponse, error)
- func (h *HarnessRegistry) RegisterHarness(ctx context.Context, req *v1.RegisterHarnessRequest) (*v1.RegisterHarnessResponse, error)
- func (h *HarnessRegistry) RetireHarness(ctx context.Context, req *v1.RetireHarnessRequest) (*v1.HarnessRegistration, error)
- type Option
- type PlacerFactory
- type Service
- func (s *Service) Approve(ctx context.Context, req *v1.ApproveRequest) (response *v1.ApproveResponse, err error)
- func (s *Service) CreateSession(ctx context.Context, req *v1.CreateSessionRequest) (session *v1.Session, err error)
- func (s *Service) DeleteSession(ctx context.Context, req *v1.DeleteSessionRequest) (session *v1.Session, err error)
- func (s *Service) Exec(req *v1.ExecRequest, stream v1.Sessions_ExecServer) (err error)
- func (s *Service) Fork(ctx context.Context, req *v1.ForkRequest) (response *v1.ForkResponse, err error)
- func (s *Service) GetSession(ctx context.Context, req *v1.GetSessionRequest) (session *v1.Session, err error)
- func (s *Service) ListSessions(ctx context.Context, req *v1.ListSessionsRequest) (resp *v1.ListSessionsResponse, err error)
- func (s *Service) Replay(req *v1.ReplayRequest, stream v1.Sessions_ReplayServer) (err error)
- func (s *Service) Resume(ctx context.Context, req *v1.ResumeRequest) (session *v1.Session, err error)
- func (s *Service) Suspend(ctx context.Context, req *v1.SuspendRequest) (session *v1.Session, err error)
Constants ¶
const MaxForkChildren = 128
MaxForkChildren bounds one fan-out. count arrives off the wire and sizes both an allocation and a provisioning loop, and each child costs a full snapshot restore on a memory-capable backend, so an unbounded value is a resource-exhaustion vector rather than a useful request. It is a guardrail, not a statement about how wide forking can scale.
Variables ¶
This section is empty.
Functions ¶
func ReserveStaticHarnessNames ¶
ReserveStaticHarnessNames reserves, in the store, every name the registry holds for a static harness, so that no host sharing the store can register one of them, now or later. It fails if one of them already has a registration, active or retired. Every host that routes sessions through a store other hosts may register harnesses in calls it before serving: otherwise a registration made by a host whose static names differ would mean a second harness under one of this host's names. NewHarnessRegistry calls it.
Service also refuses, when it creates or routes a session, a static name that has a registration row, so a host that skips this call fails closed instead of mixing the two. The reservation is what lets the registering host refuse the name up front instead.
Types ¶
type HarnessRegistry ¶
type HarnessRegistry struct {
v1.UnimplementedHarnessRegistryServer
// contains filtered or unexported fields
}
HarnessRegistry implements v1.HarnessRegistryServer over the store and placement registry the Sessions service routes through.
It is an admin surface: whoever can call RegisterHarness decides where the host sends session history and whose credential pays for model calls. agentsessionsd serves it only on an optional separate literal-loopback listener; neither that listener nor Sessions authenticates callers. Hosts that embed it must keep it off the Sessions listener and protect its reachability.
func NewHarnessRegistry ¶
func NewHarnessRegistry(store *sqlitelog.Store, registry *placement.Registry, factory PlacerFactory) (*HarnessRegistry, error)
NewHarnessRegistry builds the service, retaining every stored name (including retired ones) and resolving each usable registration to a Placer. Existing sessions on a retired harness still run. Per-registration spec, factory and descriptor failures leave only that name unservable; store enumeration/reservation errors and static-name collisions still refuse startup. It first reserves the registry's static and reserved names in the store, as ReserveStaticHarnessNames does, rather than letting one name mean two harnesses.
Registrations are loaded at startup. A registration another process makes in a shared database is served after this host restarts, or if RegisterHarness loads its identical compatible spec.
func (*HarnessRegistry) Close ¶
func (h *HarnessRegistry) Close()
Close releases every Placer the factory built. Call it after the servers that route through the registry have stopped.
func (*HarnessRegistry) GetHarness ¶
func (h *HarnessRegistry) GetHarness(ctx context.Context, req *v1.GetHarnessRequest) (*v1.HarnessRegistration, error)
GetHarness returns one static or registered harness, optionally with what it reports right now.
func (*HarnessRegistry) ListHarnesses ¶
func (h *HarnessRegistry) ListHarnesses(ctx context.Context, req *v1.ListHarnessesRequest) (*v1.ListHarnessesResponse, error)
ListHarnesses lists static and registered harnesses together in name order. Paging is keyset on the name, so a registration made while a caller pages neither skips nor repeats an entry.
func (*HarnessRegistry) RegisterHarness ¶
func (h *HarnessRegistry) RegisterHarness(ctx context.Context, req *v1.RegisterHarnessRequest) (*v1.RegisterHarnessResponse, error)
RegisterHarness stores a registration and makes it resolvable, or recognizes a repeat of one.
func (*HarnessRegistry) RetireHarness ¶
func (h *HarnessRegistry) RetireHarness(ctx context.Context, req *v1.RetireHarnessRequest) (*v1.HarnessRegistration, error)
RetireHarness refuses new sessions on a registered harness. Existing sessions keep running.
type Option ¶
type Option func(*Service)
Option configures a Service.
func WithDefaultProject ¶
WithDefaultProject sets the project assigned to sessions created without one. It must match the value given to sqlitelog.Open, so that sessions created through this service and sessions backfilled by the store migration land in the same project and a project-filtered list sees both. An empty project is never persisted: ListSessions filters on exact match, so a session stored under "" would be invisible to every real caller.
func WithLogger ¶
WithLogger enables structured operational logs.
type PlacerFactory ¶
type PlacerFactory func(name string, spec *v1.HarnessSpec) (p *placement.Placer, release func(), err error)
PlacerFactory builds the Placer for a registered harness, and a release func that frees whatever it built (nil if nothing). It is where the host decides which placements it can serve and checks a placement's address forms: an error for a new registration is FailedPrecondition and nothing is stored. At startup, an error marks only that stored registration unservable. It must not dial or provision anything: a harness that is down must not keep the host from starting.
The Placer must be built with placement.WithDescriptorID(spec.GetDescriptorId()), so that every Exec, Resume and Fork refuses a harness that reports another descriptor id. HarnessRegistry checks this and refuses a Placer that does not expect exactly that id, so a factory that drops the option fails closed instead of running turns on whatever answers.
The Placer must be new: not used, and not in any Registry. placement.Registry.Add wires it to the registry's shared session guard and checkpoint state, so its sessions cannot overlap calls through a static harness, and refuses a Placer that is in use or in another Registry.
For a RemotePlacement, runtime/remote is the backend that dials a harness already running at an address; agentsessionsd's -harness flag uses it the same way.
type Service ¶
type Service struct {
v1.UnimplementedSessionsServer
// contains filtered or unexported fields
}
Service implements v1.SessionsServer over a sqlitelog store and the controller.
func NewService ¶
NewService builds the Sessions service over store, routing each session to a harness through registry. There is no default-harness option: the registry already names its default, and a service-level override could name a harness the host does not serve, which would record a harness on every new session that no execution could ever resolve.
func (*Service) Approve ¶
func (s *Service) Approve(ctx context.Context, req *v1.ApproveRequest) (response *v1.ApproveResponse, err error)
Approve commits only the supplied decision. It deliberately does not resolve a harness: decisions remain available while compute is unavailable or the recorded route is unserved.
func (*Service) CreateSession ¶
func (s *Service) CreateSession(ctx context.Context, req *v1.CreateSessionRequest) (session *v1.Session, err error)
CreateSession persists the session's metadata and returns it. The metadata row is written before any event exists, which is what makes a freshly created session appear in ListSessions and survive a restart; compute is still provisioned lazily on first Exec.
func (*Service) DeleteSession ¶
func (s *Service) DeleteSession(ctx context.Context, req *v1.DeleteSessionRequest) (session *v1.Session, err error)
DeleteSession acknowledges permanent durable intent only. It does not resolve a harness or contact runtime resources; the retained compute projection is not a cleanup claim.
func (*Service) Exec ¶
func (s *Service) Exec(req *v1.ExecRequest, stream v1.Sessions_ExecServer) (err error)
Exec runs one turn against the session's durable log (CAS-guarded by expected_last_seq) and streams the committed LogRecords produced by the turn.
func (*Service) Fork ¶
func (s *Service) Fork(ctx context.Context, req *v1.ForkRequest) (response *v1.ForkResponse, err error)
Fork branches the session at at_seq into count children, each a new session sharing the parent's prefix chain. The whole fan-out branches from one parent checkpoint, so every child starts from identical state rather than from N separate re-derivations that can drift.
func (*Service) GetSession ¶
func (s *Service) GetSession(ctx context.Context, req *v1.GetSessionRequest) (session *v1.Session, err error)
GetSession returns the session's stored metadata and current log cursor.
func (*Service) ListSessions ¶
func (s *Service) ListSessions(ctx context.Context, req *v1.ListSessionsRequest) (resp *v1.ListSessionsResponse, err error)
ListSessions enumerates ACTIVE sessions in a project, newest first, from the store rather than from any in-memory view. The listing survives a restart and includes ACTIVE sessions that have no events and no compute yet; permanently deleted sessions are excluded.
Paging is keyset, not offset: the token carries the last (create time, uid) seen, so a session created while a caller is walking pages neither skips nor duplicates a row. An unparseable token is rejected rather than silently treated as "start over", which would otherwise show up as a caller quietly re-reading page one forever.
func (*Service) Replay ¶
func (s *Service) Replay(req *v1.ReplayRequest, stream v1.Sessions_ReplayServer) (err error)
Replay re-delivers committed records from from_seq (read-only; audit / provenance).
func (*Service) Resume ¶
func (s *Service) Resume(ctx context.Context, req *v1.ResumeRequest) (session *v1.Session, err error)
Resume restores the incarnation via the Runtime SPI, re-drives any interrupted turn, and records a RESUME marker. A pending turn's recorded name selects its exact registry entry; legacy turns and completed sessions retain the stored session default.
func (*Service) Suspend ¶
func (s *Service) Suspend(ctx context.Context, req *v1.SuspendRequest) (session *v1.Session, err error)
Suspend snapshots the incarnation and marks the session cold. The Placer records the SnapshotRef in a SUSPEND lifecycle event (§5.1) and frees the worker via the Runtime SPI.
Source Files