项目文件夹

文件
T
Asim Aslam 9fdcc24cce Implement durable execution and scoped state management for flows (#2972)
* docs: design note for flow steps + Checkpoint durable execution

* docs: fold in durable-execution decisions (State struct, single Step, run retention, retry)

* docs: rename State.Payload to State.Data

* feat(flow): ordered steps + Checkpoint durable execution

A flow can now be an ordered list of steps (a task with stages) instead
of a single LLM turn. State carries typed Data plus a Stage marker; each
step is checkpointed before and after via a pluggable Checkpoint
(store-backed by default), so a run survives a crash and resumes where it
stopped without re-running completed steps. Flow-level Retry with a
per-step override; runs retained for audit unless DeleteOnSuccess.

Step actions: Call (RPC), LLM (augmented turn), Dispatch (to an agent),
or any StepFunc. Single-step and agent-dispatch flows are unchanged.

* feat(flow): top-level re-exports + durable flow example

Expose the step/checkpoint API from the micro package (FlowSteps,
FlowStep, FlowState, FlowRetry, FlowWithCheckpoint, FlowCall/LLM/Dispatch,
Checkpoint, StoreCheckpoint) and add a runnable, key-free example
demonstrating crash + resume.

* docs: document durable flow steps (guide, README, CLI help)

* docs: blog post + changelog for durable workflows

* fix(flow): scope checkpoint keys by flow name (flow/{name}/runs/{id})

Run keys were flow/runs/{id} — a single global keyspace shared by every
flow on the default store. Namespace them by flow name so each flow's
state is kept apart. StoreCheckpoint now takes a scope argument (the flow
passes its name by default).

* feat(store): Scope handle; scope agent and flow state by name

Add store.Scope(s, database, table) — a store handle that confines every
operation to a database/table without mutating the shared store, so
co-located components don't clobber each other's table (the failure mode
of the global Init(Table(...)) approach).

Use it to keep each agent's memory and plan in its own table
(agent/{name}) and each flow's runs in its own (flow/{name}), instead of
one global table partitioned only by key prefix. Services already scope
by service name.

* feat: consistent state model — service store scoping, flow registry, list/history CLI

- service: scope store via store.Scope (database service / table name),
  retiring the Init(store.Table(name)) global-mutation hack; bridge the
  default store so handlers using store.DefaultStore stay isolated.
- flow: register in the registry as type=flow while running (with trigger
  and step count), deregister on Stop. Live discovery, like agents.
- cli: micro flow list (registry), micro flow runs <name> (durable store),
  micro agent history <name> (durable store). list = running, runs/history
  = durable, mirroring the service model.

* test: mini-universe end-to-end harness + scheduled GitHub Action

internal/harness/universe boots a small but real go-micro world — four
services, a durable checkout flow that crashes at payment and resumes,
and a guardrailed agent with a tool wrapper reached over RPC — drives the
scenario, asserts the end state (10 checks), and shuts down. Everything
is real except the LLM (mocked), so it's deterministic and needs no key;
-provider anthropic runs it live. Exits non-zero on failure, so it's an
end-to-end test, not just a demo.

Adds .github/workflows/universe.yml (push/PR/daily/dispatch) running the
universe + existing harnesses on the mock provider, plus an opt-in job
that runs live when ANTHROPIC_API_KEY is set. 'make harness' runs them
locally.

* ci: run the live universe job against AtlasCloud (ATLASCLOUD_API_KEY)

* ci: run the live universe job only on schedule or manual dispatch

The deterministic mock job still runs on push/PR/daily; the live
(AtlasCloud) job runs daily and on manual workflow_dispatch only, so
changes don't burn API credits on every PR but can still be checked
against a real model on demand.

---------

Co-authored-by: Claude <noreply@anthropic.com>
2026-06-17 12:02:47 +01:00

224 行
4.8 KiB
Go

package service
import (
"os"
"os/signal"
rtime "runtime"
"sync"
"go-micro.dev/v5/client"
"go-micro.dev/v5/cmd"
signalutil "go-micro.dev/v5/internal/util/signal"
log "go-micro.dev/v5/logger"
"go-micro.dev/v5/model"
"go-micro.dev/v5/server"
"go-micro.dev/v5/store"
)
// Service is the interface for a go-micro service.
type Service interface {
// Name returns the service name.
Name() string
// Init initializes options. Parses command line flags on first call.
Init(...Option)
// Options returns the current options.
Options() Options
// Handle registers a handler with optional server.HandlerOption args.
Handle(v interface{}, opts ...server.HandlerOption) error
// Client returns the RPC client.
Client() client.Client
// Server returns the RPC server.
Server() server.Server
// Model returns the data model backend.
Model() model.Model
// Start the service (non-blocking).
Start() error
// Stop the service.
Stop() error
// Run starts the service, blocks on signal/context, then stops.
Run() error
// String returns the implementation name.
String() string
}
type serviceImpl struct {
opts Options
once sync.Once
}
// New creates a new service with the given options.
func New(opts ...Option) Service {
return &serviceImpl{
opts: newOptions(opts...),
}
}
func (s *serviceImpl) Name() string {
return s.opts.Server.Options().Name
}
// Init initializes options. Additionally it calls cmd.Init
// which parses command line flags. cmd.Init is only called
// on first Init.
func (s *serviceImpl) Init(opts ...Option) {
// process options
for _, o := range opts {
o(&s.opts)
}
s.once.Do(func() {
// set cmd name
if len(s.opts.Cmd.App().Name) == 0 {
s.opts.Cmd.App().Name = s.Server().Options().Name
}
// Initialize the command flags, overriding new service
if err := s.opts.Cmd.Init(
cmd.Auth(&s.opts.Auth),
cmd.Broker(&s.opts.Broker),
cmd.Registry(&s.opts.Registry),
cmd.Transport(&s.opts.Transport),
cmd.Client(&s.opts.Client),
cmd.Config(&s.opts.Config),
cmd.Server(&s.opts.Server),
cmd.Store(&s.opts.Store),
cmd.Profile(&s.opts.Profile),
); err != nil {
s.opts.Logger.Log(log.FatalLevel, err)
}
// Scope the service's store to its own table (database "service",
// table = service name), consistent with how agents ("agent/{name}")
// and flows ("flow/{name}") scope their state. This replaces the
// older Init(store.Table(name)) global mutation with a composable
// scoped handle: each service gets an isolated handle that works
// even when several run in one process. When the service uses the
// package default store, bridge it to the same scope so handlers
// that reach for store.DefaultStore stay isolated too.
name := s.opts.Cmd.App().Name
wasDefault := s.opts.Store == store.DefaultStore
s.opts.Store = store.Scope(s.opts.Store, "service", name)
if wasDefault {
store.DefaultStore = s.opts.Store
}
})
}
func (s *serviceImpl) Options() Options {
return s.opts
}
func (s *serviceImpl) Client() client.Client {
return s.opts.Client
}
func (s *serviceImpl) Server() server.Server {
return s.opts.Server
}
func (s *serviceImpl) Model() model.Model {
return s.opts.Model
}
func (s *serviceImpl) String() string {
return "micro"
}
func (s *serviceImpl) Start() error {
for _, fn := range s.opts.BeforeStart {
if err := fn(); err != nil {
return err
}
}
if err := s.opts.Server.Start(); err != nil {
return err
}
for _, fn := range s.opts.AfterStart {
if err := fn(); err != nil {
return err
}
}
return nil
}
func (s *serviceImpl) Stop() error {
var gerr error
for _, fn := range s.opts.BeforeStop {
if err := fn(); err != nil {
gerr = err
}
}
if err := s.opts.Server.Stop(); err != nil {
return err
}
for _, fn := range s.opts.AfterStop {
if err := fn(); err != nil {
gerr = err
}
}
return gerr
}
func (s *serviceImpl) Handle(v interface{}, opts ...server.HandlerOption) error {
return s.opts.Server.Handle(
s.opts.Server.NewHandler(v, opts...),
)
}
func (s *serviceImpl) Run() (err error) {
logger := s.opts.Logger
// exit when help flag is provided
for _, v := range os.Args[1:] {
if v == "-h" || v == "--help" {
os.Exit(0)
}
}
// start the profiler
if s.opts.Profile != nil {
// to view mutex contention
rtime.SetMutexProfileFraction(5)
// to view blocking profile
rtime.SetBlockProfileRate(1)
if err = s.opts.Profile.Start(); err != nil {
return err
}
defer func() {
if nerr := s.opts.Profile.Stop(); nerr != nil {
logger.Log(log.ErrorLevel, nerr)
}
}()
}
logger.Logf(log.InfoLevel, "Starting [service] %s", s.Name())
if err = s.Start(); err != nil {
return err
}
ch := make(chan os.Signal, 1)
if s.opts.Signal {
signal.Notify(ch, signalutil.Shutdown()...)
}
select {
// wait on kill signal
case <-ch:
// wait on context cancel
case <-s.opts.Context.Done():
}
return s.Stop()
}