Documentation
¶
Overview ¶
Package filesystem is a single-process, crash-atomic reference implementation of agentkit.Repository backed by a single JSON snapshot file. It holds the same in-memory state as the memory implementation and, on every write, atomically replaces state.json via temp-file + fsync + rename + directory fsync (the rename is the commit point, so no WAL is needed).
Constraints (stated honestly): it is single-process only — a LOCK file (flock) makes a second New on the same directory fail — and it rewrites the whole snapshot on every write, so I/O is O(state size). It suits development, tests, and small one-shot runs; large or high-throughput deployments should use a database-backed Repository. Serve's WithPollConcurrency (in-process goroutine concurrency) is supported.
Index ¶
- type Repository
- func (r *Repository) Apply(ctx context.Context, cs agentkit.ChangeSet) error
- func (r *Repository) ClaimNextProcess(ctx context.Context, workerID string, leaseUntil time.Time, now time.Time) (*agentkit.Process, error)
- func (r *Repository) Close() error
- func (r *Repository) FindOpenProcessBySubject(ctx context.Context, subject agentkit.SubjectRef) (*agentkit.Process, error)
- func (r *Repository) FindProcessByIdempotencyKey(ctx context.Context, key string) (*agentkit.Process, error)
- func (r *Repository) GetProcess(ctx context.Context, pid agentkit.ProcessID) (*agentkit.Process, error)
- func (r *Repository) ListAwaits(ctx context.Context, pid agentkit.ProcessID) ([]*agentkit.Await, error)
- func (r *Repository) ListEvents(ctx context.Context, pid agentkit.ProcessID, q agentkit.EventQuery) ([]*agentkit.Event, error)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Repository ¶
type Repository struct {
// contains filtered or unexported fields
}
Repository is the filesystem-backed agentkit.Repository.
func New ¶
func New(dir string) (*Repository, error)
New opens (or creates) a filesystem Repository rooted at dir. It acquires an exclusive flock on {dir}/LOCK (a second concurrent New fails), removes any stray uncommitted state.json.tmp, and loads state.json if present.
func (*Repository) Apply ¶
Apply applies a ChangeSet atomically and persists it. On any precondition failure it writes nothing and returns agentkit.ErrConflict.
func (*Repository) ClaimNextProcess ¶
func (r *Repository) ClaimNextProcess(ctx context.Context, workerID string, leaseUntil time.Time, now time.Time) (*agentkit.Process, error)
ClaimNextProcess atomically claims one runnable Process and persists the claim. No target -> (nil, nil).
func (*Repository) Close ¶
func (r *Repository) Close() error
Close releases the flock. After Close the Repository must not be used; reopening with New on the same directory reloads the persisted state.
func (*Repository) FindOpenProcessBySubject ¶
func (r *Repository) FindOpenProcessBySubject(ctx context.Context, subject agentkit.SubjectRef) (*agentkit.Process, error)
FindOpenProcessBySubject finds an open Process holding the subject.
func (*Repository) FindProcessByIdempotencyKey ¶
func (r *Repository) FindProcessByIdempotencyKey(ctx context.Context, key string) (*agentkit.Process, error)
FindProcessByIdempotencyKey finds a Process by idempotency key.
func (*Repository) GetProcess ¶
func (r *Repository) GetProcess(ctx context.Context, pid agentkit.ProcessID) (*agentkit.Process, error)
GetProcess returns the Process, or agentkit.ErrProcessNotFound.
func (*Repository) ListAwaits ¶
func (r *Repository) ListAwaits(ctx context.Context, pid agentkit.ProcessID) ([]*agentkit.Await, error)
ListAwaits returns all awaits of a Process.
func (*Repository) ListEvents ¶
func (r *Repository) ListEvents(ctx context.Context, pid agentkit.ProcessID, q agentkit.EventQuery) ([]*agentkit.Event, error)
ListEvents returns a Process's events in append order, after the cursor and capped at limit.
Source Files