diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 02e5128..f06f8cb 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -97,6 +97,11 @@ The durable custom-tool gate scenario can be selected with its documented example aligned with the complete-barrier, partial-result, duplicate-result, worker-replacement, and interactive live-model assertions. +The specialist-team user journey runs with +`scripts/with-dev-env make demo-multi-agent-team`. Keep it aligned with the +ordinary-child, Advisor, real-usage, completion-barrier, and persistent-follow-up +contracts covered by the multi-agent service and runtime tests. + Cookbook-derived examples describe real Mango user journeys, not probe-only demos. Keep an offline deterministic test for exact runtime and recovery invariants, and a runnable public-API example for the documented live-model diff --git a/Makefile b/Makefile index 88cc144..de0e86c 100644 --- a/Makefile +++ b/Makefile @@ -21,6 +21,7 @@ MANGO_TEST_S3_BUCKET ?= mango-test MANGO_TEST_S3_ACCESS_KEY ?= minioadmin MANGO_TEST_S3_SECRET_KEY ?= minioadmin MANGO_EXAMPLE_MODEL_ID ?= $(MANGO_MODEL_ID) +MANGO_EXAMPLE_ADVISOR_MODEL_ID ?= $(MANGO_EXAMPLE_MODEL_ID) DOCKER_BUILD_ARGS := --build-arg VERSION=$(VERSION) --build-arg REVISION=$(REVISION) ifneq ($(strip $(GOPROXY)),) @@ -31,6 +32,7 @@ endif .PHONY: help build lint test test-race test-service test-model-live test-platform-live \ test-coding-agent test-coding-agent-live test-hitl-gate demo-hitl-gate \ + demo-multi-agent-team \ vet verify security docs-check image image-smoke dev-env-init \ local-config local-up local-down local-health local-ps local-logs @@ -47,6 +49,7 @@ help: @echo " make test-coding-agent-live run the iterate scenario against the live model" @echo " make test-hitl-gate run the durable custom-tool HITL scenario" @echo " make demo-hitl-gate run the interactive HITL example over public HTTP" + @echo " make demo-multi-agent-team run the interactive multi-agent example over public HTTP" @echo " make vet run go vet" @echo " make verify run the core Go checks" @echo " make security scan reachable Go code and high-severity npm issues" @@ -122,6 +125,12 @@ demo-hitl-gate: env -u MANGO_MODEL_BASE_URL -u MANGO_MODEL_API_KEY -u MANGO_MODEL_AUTH -u MANGO_MODEL_ID \ $(GO) run ./examples/hitl-gate +demo-multi-agent-team: + MANGO_EXAMPLE_MODEL_ID='$(MANGO_EXAMPLE_MODEL_ID)' \ + MANGO_EXAMPLE_ADVISOR_MODEL_ID='$(MANGO_EXAMPLE_ADVISOR_MODEL_ID)' \ + env -u MANGO_MODEL_BASE_URL -u MANGO_MODEL_API_KEY -u MANGO_MODEL_AUTH -u MANGO_MODEL_ID \ + $(GO) run ./examples/multi-agent-team + vet: $(GO) vet ./... diff --git a/docs/capabilities.md b/docs/capabilities.md index c0e13c7..356d8fd 100644 --- a/docs/capabilities.md +++ b/docs/capabilities.md @@ -51,7 +51,7 @@ and service test suites. | [Vaults](api/vaults.md) | Limited | Encrypted Vault/Credential lifecycle, ordered Session attachment, OAuth validation, expiry refresh, and token rotation. Environment-variable egress and refresh-failure notifications are not implemented. | | [Deployments](api/deployments.md) | Limited | Deployment/Run lifecycle, pinned Agent Versions, Session budget templates, manual runs, cron scheduling, leases, and atomic success/failure records. Repository resources and Agent-archive propagation remain open. | | [Environment Work](api/environment-work.md) | Limited | Self-hosted worker leases, polling, heartbeats, state transitions, reclaim, and Session activation. Environment-key issuance, tenant-scoped authorization, Work secrets, and health-check Work remain open. | -| [Multi-agent](guides/multi-agent.md) | Limited | Persistent ordinary child Agents plus primary-only Mango-managed Advisor consultations over client tool calls, independent transcripts/events/usage, shared Session budgets, reports, routing, interrupts, retries, archive, deletion, and durable context-compaction checkpoints. Broader live-provider evidence and targeted interruption timing remain open. | +| [Multi-agent](guides/multi-agent.md) | Limited | Persistent ordinary child Agents plus primary-only Mango-managed Advisor consultations over client tool calls, independent transcripts/events/usage, shared Session budgets, reports, routing, interrupts, retries, archive, deletion, and durable context-compaction checkpoints. A real-provider specialist-team journey covers parallel delegation, Advisor consultation, synthesis, and persistent-Thread follow-up; repeated broader live-provider evidence and targeted interruption timing remain open. | | [Sandbox adapters](sandboxes.md) | Limited / Preview | Local and Docker are available. E2B, CubeSandbox, OpenSandbox, and Daytona have durable bindings, materialize File Resources and custom Skills, and publish Session Outputs through their official SDKs. Remote resource copies have documented limitations; E2B/Cube additionally buffer file transfers. Remote adapters remain Preview pending repeated live conformance and production routing policy. | | Distributed operation | Limited | API and worker roles scale independently around PostgreSQL, Temporal, and NATS. Worker Versioning, heterogeneous-provider routing, distributed Files reconciliation, and production rollout evidence remain open. | diff --git a/docs/examples/multi-agent-team.md b/docs/examples/multi-agent-team.md new file mode 100644 index 0000000..21fd429 --- /dev/null +++ b/docs/examples/multi-agent-team.md @@ -0,0 +1,86 @@ +--- +title: Coordinate a specialist team +slug: /examples/multi-agent-team +--- + +# Coordinate a specialist team + +This example runs a release-readiness review through Mango's public HTTP API. +A coordinator delegates independent work to reliability and security Agents, +consults an Advisor, waits for all three responses, and produces a decision. A +second interactive turn adds a new constraint and must continue the existing +reliability Thread instead of creating another one. + +Every inference in the documented run uses the real model endpoint configured +on the Mango worker. The example client has no model-provider credentials and +does not invoke a hosted agent service. It also uses no simulated GitHub, web, +database, or other third-party boundary: the workflow exercises only Mango's +Agents, Session, Events, and Session Threads. + +## What the run proves + +The program checks observable product behavior rather than matching exact model +wording: + +1. The coordinator starts exactly one `reliability_reviewer` Thread and one + `security_reviewer` Thread. +2. Both specialists return non-empty reports with real provider token usage. +3. The coordinator does not produce its decision until both specialist reports + and the configured Advisor's challenge are available. The model may consult + the Advisor concurrently with the specialists. +4. The Advisor appears as one automatically terminated + `{"type":"advisor"}` Session Thread with its own usage. +5. The terminal user supplies a follow-up constraint. +6. The coordinator addresses the existing reliability Thread with + `session_thread_id` and waits for its new report. +7. The follow-up increases usage on that same Thread without creating another + specialist or Advisor Thread. + +Mango's ordinary offline service tests cover the deterministic delegation, +Advisor, persistence, retry, interrupt, and follow-up invariants. This example +adds the explicitly opt-in real-model evidence that public CI cannot provide. + +## Run the example + +Configure `MANGO_MODEL_*` in `~/.config/mango/dev.env` as described in +[Getting started](../getting-started.md). Start the local stack with that +environment so its worker uses the real Messages-shaped endpoint: + +```bash +scripts/with-dev-env make local-up +make local-health +``` + +Then run the public-HTTP client: + +```bash +scripts/with-dev-env make demo-multi-agent-team +``` + +The Make target passes the configured model IDs to the example but removes the +provider base URL and key from the client process. Set +`MANGO_EXAMPLE_ADVISOR_MODEL_ID` when the Advisor should use a different model; +otherwise it uses `MANGO_EXAMPLE_MODEL_ID`. + +After the first decision, the terminal asks for another release constraint. +Enter one or press Return to use the displayed default. The real coordinator +must route it through the persistent reliability Thread before returning a +revised decision. + +Set `MANGO_EXAMPLE_KEEP_RESOURCES=1` to retain the Session, three Agent +resources, and Environment for inspection. Otherwise the program deletes the +Session and Environment and archives the Agents after the verification. + +## Design boundary + +The specialist-team user problem is informed by Anthropic's public +[`CMA_coordinate_specialist_team` cookbook](https://github.com/anthropics/claude-cookbooks/blob/main/managed_agents/CMA_coordinate_specialist_team.ipynb). +Mango adopts the useful coordinator, scoped specialist, persistent Thread, and +Advisor workflow. It replaces the hosted data, web-search, SDK, and +`send_to_parent` presentation with synthetic release facts, Mango's public HTTP +API, and runtime-owned child completion reports. + +The example does not define Mango's multi-agent contract and adds no +scenario-specific runtime behavior. See +[Run a multi-agent Session](../guides/multi-agent.md) for the reusable workflow +and [Session Threads](../api/session-threads.md) for the HTTP contract. diff --git a/docs/getting-started.md b/docs/getting-started.md index 73ce892..6498ffc 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -194,18 +194,25 @@ scripts/with-dev-env make test-coding-agent-live # Runs the public-HTTP expense gate example. The real model generates the # decide/escalate calls and the terminal prompts for the human decision. scripts/with-dev-env make demo-hitl-gate + +# Runs a real coordinator, two specialist Agents, one Advisor consultation, +# and an interactive follow-up on a persistent child Thread. +scripts/with-dev-env make demo-multi-agent-team ``` The [coding-agent iteration example](examples/coding-agent-iterate.md) explains the corresponding user workflow and the Mango resources involved. It is a design walkthrough rather than a second test runner. The [HITL gate example](examples/hitl-gate.md) documents the interactive public-HTTP -example and its application-owned action boundary. +example and its application-owned action boundary. The +[specialist-team example](examples/multi-agent-team.md) verifies real-model +delegation, Advisor usage, and persistent Thread follow-up. These commands never print the API key. The model-only smoke test does not enable tools; the platform tests use an isolated Docker sandbox, the coding scenario excludes Web Search/Fetch from its least-privilege toolset, and the -HITL example removes provider credentials from its client process. Live checks +interactive examples remove provider credentials from their client processes. +Live checks are excluded from public CI because external credentials, availability, latency, user input, and cost are not deterministic. Use a newly issued key if a credential has ever appeared in chat, logs, or shell diff --git a/docs/guides/multi-agent.md b/docs/guides/multi-agent.md index 601f7cb..f8ff17e 100644 --- a/docs/guides/multi-agent.md +++ b/docs/guides/multi-agent.md @@ -13,6 +13,10 @@ Multi-agent execution uses the ordinary Session and Event APIs. Delegation is not a client-side endpoint: a coordinator receives the private `list_agents` and `send_to_agent` model tools and decides when to call them. +For a runnable public-HTTP scenario with a real coordinator, two specialists, +an Advisor, and persistent follow-up, see +[Coordinate a specialist team](../examples/multi-agent-team.md). + ## Prerequisites - Complete [Getting started](../getting-started.md). diff --git a/docs/provenance.md b/docs/provenance.md index e42724d..1bee070 100644 --- a/docs/provenance.md +++ b/docs/provenance.md @@ -168,6 +168,31 @@ support, so they run the same offline and opt-in live conformance suites. assumptions. Durable outbound webhook delivery remains separate work with its own signing, retry, idempotency, and observability requirements. +## Multi-agent specialist team + +- Anthropic's public + [`CMA_coordinate_specialist_team` cookbook](https://github.com/anthropics/claude-cookbooks/blob/main/managed_agents/CMA_coordinate_specialist_team.ipynb) + supplied the useful specialist-team user problem: a coordinator delegates + role-scoped work, waits for reports, consults an Advisor, and synthesizes a + final decision. +- Mango's real-model example adopts that high-level workflow but uses synthetic + release-readiness facts and no Web, hosted data, or third-party integration. + The client exercises Mango's public HTTP resources and inspects its persisted + Event and Session Thread projections. +- Mango changes child completion semantics deliberately. Ordinary child Agents + finish a turn and the runtime projects their report to the coordinator; they + do not receive or need a hosted `send_to_parent` tool. Persistent follow-up is + addressed through Mango's runtime-owned `send_to_agent` tool and the existing + `session_thread_id`. +- The scenario verifies one real provider run with two ordinary children, a + primary-only Advisor consultation, per-Thread usage, a final synthesis + barrier, and an interactive follow-up. Deterministic service tests remain + authoritative for retry, interruption, recovery, archive, and deletion + invariants. +- Mango did not adopt the Cookbook's SDK calls, cloud Environment fields, + bundled sales collateral, web-search dependency, hosted model restrictions, + or exact response text. + ## Custom Skills - The public [Claude Managed Agents Skills guide](https://platform.claude.com/docs/en/managed-agents/skills) diff --git a/examples/multi-agent-team/main.go b/examples/multi-agent-team/main.go new file mode 100644 index 0000000..a70584f --- /dev/null +++ b/examples/multi-agent-team/main.go @@ -0,0 +1,708 @@ +// Command multi-agent-team runs a real-model coordinator, specialist, Advisor, +// and persistent follow-up workflow against Mango's public HTTP API. It uses +// only net/http so the example does not depend on an unpublished Mango SDK. +package main + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "os" + "os/signal" + "sort" + "strings" + "syscall" + "time" +) + +const ( + defaultBaseURL = "http://localhost:8080" + defaultAPIKey = "sk-mango-local-development" + requestTimeout = 30 * time.Second + scenarioTimeout = 12 * time.Minute +) + +type apiClient struct { + baseURL string + apiKey string + http *http.Client +} + +type resource struct { + ID string `json:"id"` + Version int `json:"version"` +} + +type eventList struct { + Data []event `json:"data"` +} + +type event struct { + ID string `json:"id"` + Type string `json:"type"` + AgentName string `json:"agent_name"` + SessionThreadID string `json:"session_thread_id"` + FromAgentName string `json:"from_agent_name"` + FromSessionThreadID string `json:"from_session_thread_id"` + Content []contentBlock `json:"content"` + StopReason *stopReason `json:"stop_reason"` + Error any `json:"error"` +} + +type contentBlock struct { + Type string `json:"type"` + Text string `json:"text"` +} + +type stopReason struct { + Type string `json:"type"` +} + +type sessionThreadList struct { + Data []sessionThread `json:"data"` +} + +type sessionThread struct { + ID string `json:"id"` + ParentThreadID *string `json:"parent_thread_id"` + Agent threadAgent `json:"agent"` + Status string `json:"status"` + Usage threadUsage `json:"usage"` +} + +type threadAgent struct { + Type string `json:"type"` + Name string `json:"name"` +} + +type threadUsage struct { + InputTokens int64 `json:"input_tokens"` + OutputTokens int64 `json:"output_tokens"` +} + +type roundResult struct { + Events []event + Answer string +} + +type createdResources struct { + EnvironmentID string + AgentIDs []string + SessionID string +} + +func main() { + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "multi-agent example failed:", err) + os.Exit(1) + } +} + +func run() error { + modelID := strings.TrimSpace(os.Getenv("MANGO_EXAMPLE_MODEL_ID")) + if modelID == "" { + return errors.New("MANGO_EXAMPLE_MODEL_ID is required; set it to the model configured on the Mango worker") + } + advisorModelID := strings.TrimSpace(os.Getenv("MANGO_EXAMPLE_ADVISOR_MODEL_ID")) + if advisorModelID == "" { + advisorModelID = modelID + } + baseURL := strings.TrimSpace(os.Getenv("MANGO_EXAMPLE_BASE_URL")) + if baseURL == "" { + baseURL = defaultBaseURL + } + apiKey := strings.TrimSpace(os.Getenv("MANGO_API_KEY")) + if apiKey == "" { + apiKey = defaultAPIKey + } + keepResources := os.Getenv("MANGO_EXAMPLE_KEEP_RESOURCES") == "1" + + signalContext, stop := signal.NotifyContext( + context.Background(), os.Interrupt, syscall.SIGTERM, + ) + defer stop() + ctx, cancel := context.WithTimeout(signalContext, scenarioTimeout) + defer cancel() + + client := &apiClient{ + baseURL: strings.TrimRight(baseURL, "/"), + apiKey: apiKey, + http: &http.Client{Timeout: requestTimeout}, + } + if err := client.get(ctx, "/readyz", nil); err != nil { + return fmt.Errorf("mango is not ready at %s: %w", client.baseURL, err) + } + + resources := createdResources{} + succeeded := false + defer func() { + if keepResources || (!succeeded && resources.SessionID != "") { + if !succeeded && resources.SessionID != "" { + fmt.Fprintf( + os.Stderr, + "Keeping failed Session %s for inspection; delete it after diagnosis.\n", + resources.SessionID, + ) + } + return + } + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), requestTimeout) + defer cleanupCancel() + client.cleanup(cleanupCtx, resources) + }() + + fmt.Println("Creating two specialists, a coordinator, and a Session through the Mango HTTP API...") + environment, err := client.create(ctx, "/v1/environments", map[string]any{ + "name": "Multi-agent release review example", + "config": map[string]any{"type": "cloud"}, + }) + if err != nil { + return fmt.Errorf("create Environment: %w", err) + } + resources.EnvironmentID = environment.ID + + reliability, err := client.create(ctx, "/v1/agents", map[string]any{ + "name": "reliability_reviewer", + "description": "Reviews rollout safety, failure modes, observability, and rollback plans.", + "model": modelID, + "system": strings.Join([]string{ + "You are the reliability specialist on a release-review team.", + "Analyze only the task sent by the coordinator.", + "Return a concise report beginning with RELIABILITY_REPORT: and include release blockers, rollback conditions, and the smallest safe rollout.", + "Do not address the end user and do not claim to have changed any system.", + }, " "), + }) + if err != nil { + return fmt.Errorf("create reliability Agent: %w", err) + } + resources.AgentIDs = append(resources.AgentIDs, reliability.ID) + + security, err := client.create(ctx, "/v1/agents", map[string]any{ + "name": "security_reviewer", + "description": "Reviews authentication, secret handling, auditability, and abuse paths.", + "model": modelID, + "system": strings.Join([]string{ + "You are the security specialist on a release-review team.", + "Analyze only the task sent by the coordinator.", + "Return a concise report beginning with SECURITY_REPORT: and include threat assumptions, credential risks, audit requirements, and release blockers.", + "Do not address the end user and do not claim to have changed any system.", + }, " "), + }) + if err != nil { + return fmt.Errorf("create security Agent: %w", err) + } + resources.AgentIDs = append(resources.AgentIDs, security.ID) + + coordinator, err := client.create(ctx, "/v1/agents", map[string]any{ + "name": "release_review_coordinator", + "model": modelID, + "system": strings.Join([]string{ + "You coordinate release-readiness reviews and must use the team rather than doing specialist analysis yourself.", + "For the first user request, immediately send one self-contained task to reliability_reviewer and one to security_reviewer; these independent tasks may run in parallel.", + "Call advisor exactly once for an independent challenge while the specialists run or after they report. Return one final decision only after both specialist reports and the advice arrive, with headings Decision, Reliability, Security, Advisor challenge, and Next steps.", + "Do not invent reports and do not give the user a final decision while required team work is still running.", + "For a later message beginning FOLLOW-UP:, call list_agents, find the existing reliability_reviewer Thread, and send the new constraint to that exact Thread using session_thread_id.", + "Do not start a second reliability Thread and do not consult the advisor again for the follow-up. Wait for the follow-up report, then revise the decision.", + }, " "), + "multiagent": map[string]any{ + "type": "coordinator", + "agents": []any{ + map[string]any{ + "type": "agent", "id": reliability.ID, + "version": reliability.Version, + }, + map[string]any{ + "type": "agent", "id": security.ID, + "version": security.Version, + }, + map[string]any{"type": "advisor", "model": advisorModelID}, + }, + }, + }) + if err != nil { + return fmt.Errorf("create coordinator Agent: %w", err) + } + resources.AgentIDs = append(resources.AgentIDs, coordinator.ID) + + session, err := client.create(ctx, "/v1/sessions", map[string]any{ + "agent": map[string]any{ + "type": "agent", "id": coordinator.ID, + "version": coordinator.Version, + }, + "environment_id": environment.ID, + "title": "Release readiness team review", + }) + if err != nil { + return fmt.Errorf("create Session: %w", err) + } + resources.SessionID = session.ID + fmt.Printf( + "Session %s is using worker model %s and Advisor model %s.\n", + session.ID, modelID, advisorModelID, + ) + + initialHistory, err := client.events(ctx, session.ID) + if err != nil { + return fmt.Errorf("read initial Session history: %w", err) + } + initialPrompt := strings.Join([]string{ + "Review whether we should release a payments API migration from static API keys to 15-minute access tokens tomorrow.", + "Facts: 15 client applications are in scope; a 5% canary is available; rollback restores the previous gateway in about 20 minutes; audit logs exclude token values; one legacy client can rotate credentials only by taking a two-hour outage; there is no automatic client-side fallback.", + "Use the required specialist-team process and give a go, conditional-go, or no-go decision.", + }, " ") + if err := client.sendUserMessage(ctx, session.ID, initialPrompt); err != nil { + return fmt.Errorf("send initial user message: %w", err) + } + + fmt.Println("Waiting for both real specialist models and the real Advisor consultation...") + initialRound, err := client.waitForRound( + ctx, + session.ID, + len(initialHistory), + []string{ + "reliability_reviewer", + "security_reviewer", + "anthropic.advisor", + }, + ) + if err != nil { + return fmt.Errorf("initial team review: %w", err) + } + if err := validateInitialRound(initialRound.Events); err != nil { + return err + } + initialThreads, err := client.threads(ctx, session.ID) + if err != nil { + return fmt.Errorf("list Session Threads: %w", err) + } + threadIDs, initialUsage, err := validateInitialThreads(initialThreads) + if err != nil { + return err + } + + printReports(initialRound.Events) + fmt.Println("\nCoordinator decision:") + fmt.Println(initialRound.Answer) + + reader := bufio.NewReader(os.Stdin) + defaultFollowUp := "Assume rollback now takes 45 minutes and the legacy client has no maintenance window this week. Revisit the recommendation." + fmt.Println("\nAdd a release constraint for the existing reliability specialist.") + fmt.Printf("Follow-up [%s]: ", defaultFollowUp) + line, readErr := reader.ReadString('\n') + followUp := strings.TrimSpace(line) + if followUp == "" { + followUp = defaultFollowUp + } + if readErr != nil && !errors.Is(readErr, io.EOF) { + return fmt.Errorf("read follow-up: %w", readErr) + } + + followHistory, err := client.events(ctx, session.ID) + if err != nil { + return fmt.Errorf("read history before follow-up: %w", err) + } + if err := client.sendUserMessage( + ctx, + session.ID, + "FOLLOW-UP: "+followUp, + ); err != nil { + return fmt.Errorf("send follow-up user message: %w", err) + } + + fmt.Println("Waiting for a follow-up on the existing reliability Thread...") + followRound, err := client.waitForRound( + ctx, + session.ID, + len(followHistory), + []string{"reliability_reviewer"}, + ) + if err != nil { + return fmt.Errorf("follow-up review: %w", err) + } + if err := validateFollowUpRound( + followRound.Events, + threadIDs["reliability_reviewer"], + ); err != nil { + return err + } + followThreads, err := client.threads(ctx, session.ID) + if err != nil { + return fmt.Errorf("list Threads after follow-up: %w", err) + } + if err := validateFollowUpThreads( + followThreads, + threadIDs, + initialUsage, + ); err != nil { + return err + } + + printReports(followRound.Events) + fmt.Println("\nRevised coordinator decision:") + fmt.Println(followRound.Answer) + fmt.Println("\nVerified: two specialist Threads, one Advisor consultation, real usage, and persistent reliability follow-up.") + + succeeded = true + if keepResources { + fmt.Printf("\nKeeping Session %s for inspection.\n", session.ID) + fmt.Printf("History: GET /v1/sessions/%s/events\n", session.ID) + fmt.Printf("Threads: GET /v1/sessions/%s/threads\n", session.ID) + return nil + } + fmt.Println("\nSet MANGO_EXAMPLE_KEEP_RESOURCES=1 on a later run to inspect its durable history.") + return nil +} + +func (c *apiClient) waitForRound( + ctx context.Context, + sessionID string, + start int, + requiredReports []string, +) (roundResult, error) { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + for { + history, err := c.events(ctx, sessionID) + if err != nil { + return roundResult{}, fmt.Errorf("list events: %w", err) + } + if len(history) < start { + return roundResult{}, errors.New("session event history moved backwards") + } + round := history[start:] + if failure := firstEvent(round, "session.error"); failure != nil { + encoded, _ := json.Marshal(failure) + return roundResult{}, fmt.Errorf("session failed: %s", encoded) + } + idle := lastEvent(round, "session.status_idle") + if idle != nil && idle.StopReason != nil && idle.StopReason.Type == "end_turn" { + for _, name := range requiredReports { + report := lastReport(round, name) + if report == nil { + return roundResult{}, fmt.Errorf("session reached end_turn without a report from %s", name) + } + if strings.TrimSpace(eventText(*report)) == "" { + return roundResult{}, fmt.Errorf("session reached end_turn with an empty report from %s", name) + } + } + message := lastEvent(round, "agent.message") + if message == nil || strings.TrimSpace(eventText(*message)) == "" { + return roundResult{}, errors.New("session ended without a non-empty coordinator message") + } + return roundResult{Events: round, Answer: strings.TrimSpace(eventText(*message))}, nil + } + select { + case <-ctx.Done(): + return roundResult{}, fmt.Errorf("wait for end_turn: %w", ctx.Err()) + case <-ticker.C: + } + } +} + +func validateInitialRound(events []event) error { + created := map[string]int{} + for _, item := range events { + switch item.Type { + case "session.thread_created": + created[item.AgentName]++ + } + } + for _, name := range []string{ + "reliability_reviewer", "security_reviewer", "anthropic.advisor", + } { + if created[name] != 1 { + return fmt.Errorf("thread_created count for %s = %d, want 1", name, created[name]) + } + } + if lastReport(events, "anthropic.advisor") == nil { + return errors.New("advisor report was not projected onto the primary event history") + } + answer := eventText(*lastEvent(events, "agent.message")) + for _, term := range []string{"reliab", "security", "advisor"} { + if !strings.Contains(strings.ToLower(answer), term) { + return fmt.Errorf("coordinator answer did not incorporate %s evidence", term) + } + } + return nil +} + +func validateInitialThreads( + threads []sessionThread, +) (map[string]string, map[string]int64, error) { + if len(threads) != 4 { + return nil, nil, fmt.Errorf("session has %d Threads, want primary, two specialists, and one Advisor", len(threads)) + } + threadIDs := map[string]string{} + usage := map[string]int64{} + advisorCount := 0 + for _, thread := range threads { + label := thread.Agent.Name + if thread.Agent.Type == "advisor" { + label = "anthropic.advisor" + advisorCount++ + if thread.Status != "terminated" { + return nil, nil, fmt.Errorf( + "advisor Thread status = %s, want terminated", thread.Status, + ) + } + } else if thread.Status != "idle" { + return nil, nil, fmt.Errorf( + "agent Thread %s status = %s, want idle", thread.ID, thread.Status, + ) + } + tokens := thread.Usage.InputTokens + thread.Usage.OutputTokens + if tokens <= 0 { + return nil, nil, fmt.Errorf("thread %s (%s) has no provider token usage", thread.ID, label) + } + usage[label] = tokens + if thread.ParentThreadID != nil && thread.Agent.Type != "advisor" { + if _, exists := threadIDs[label]; exists { + return nil, nil, fmt.Errorf("more than one child Thread exists for %s", label) + } + threadIDs[label] = thread.ID + } + } + if advisorCount != 1 { + return nil, nil, fmt.Errorf("advisor Thread count = %d, want 1", advisorCount) + } + for _, name := range []string{"reliability_reviewer", "security_reviewer"} { + if threadIDs[name] == "" { + return nil, nil, fmt.Errorf("missing child Thread for %s", name) + } + } + return threadIDs, usage, nil +} + +func validateFollowUpRound(events []event, reliabilityThreadID string) error { + if created := firstEvent(events, "session.thread_created"); created != nil { + return fmt.Errorf("follow-up created an unexpected new Thread for %s", created.AgentName) + } + if advisor := lastReport(events, "anthropic.advisor"); advisor != nil { + return errors.New("follow-up made an unexpected second Advisor consultation") + } + report := lastReport(events, "reliability_reviewer") + if report == nil { + return errors.New("follow-up returned no reliability report") + } + if report.FromSessionThreadID != reliabilityThreadID { + return fmt.Errorf( + "follow-up report came from Thread %s, want persistent Thread %s", + report.FromSessionThreadID, + reliabilityThreadID, + ) + } + return nil +} + +func validateFollowUpThreads( + threads []sessionThread, + initialThreadIDs map[string]string, + initialUsage map[string]int64, +) error { + if len(threads) != 4 { + return fmt.Errorf("follow-up changed Thread count to %d, want 4", len(threads)) + } + reliabilityCount := 0 + advisorCount := 0 + for _, thread := range threads { + if thread.Agent.Type == "advisor" { + advisorCount++ + if thread.Status != "terminated" { + return fmt.Errorf( + "advisor Thread status after follow-up = %s, want terminated", + thread.Status, + ) + } + continue + } + if thread.ParentThreadID == nil || thread.Agent.Name != "reliability_reviewer" { + continue + } + reliabilityCount++ + if thread.Status != "idle" { + return fmt.Errorf( + "reliability Thread status after follow-up = %s, want idle", + thread.Status, + ) + } + if thread.ID != initialThreadIDs["reliability_reviewer"] { + return fmt.Errorf("follow-up replaced reliability Thread %s with %s", initialThreadIDs["reliability_reviewer"], thread.ID) + } + tokens := thread.Usage.InputTokens + thread.Usage.OutputTokens + if tokens <= initialUsage["reliability_reviewer"] { + return fmt.Errorf("persistent reliability Thread usage did not increase: %d <= %d", tokens, initialUsage["reliability_reviewer"]) + } + } + if reliabilityCount != 1 { + return fmt.Errorf("reliability child Thread count = %d, want 1", reliabilityCount) + } + if advisorCount != 1 { + return fmt.Errorf("advisor Thread count after follow-up = %d, want 1", advisorCount) + } + return nil +} + +func printReports(events []event) { + reports := make([]event, 0, 3) + for _, item := range events { + if item.Type == "agent.thread_message_received" { + reports = append(reports, item) + } + } + sort.SliceStable(reports, func(i, j int) bool { + return reports[i].FromAgentName < reports[j].FromAgentName + }) + for _, report := range reports { + fmt.Printf("\n[%s, Thread %s]\n%s\n", report.FromAgentName, report.FromSessionThreadID, strings.TrimSpace(eventText(report))) + } +} + +func eventText(item event) string { + parts := make([]string, 0, len(item.Content)) + for _, block := range item.Content { + if block.Type == "text" && strings.TrimSpace(block.Text) != "" { + parts = append(parts, strings.TrimSpace(block.Text)) + } + } + return strings.Join(parts, "\n") +} + +func firstEvent(events []event, eventType string) *event { + for index := range events { + if events[index].Type == eventType { + return &events[index] + } + } + return nil +} + +func lastEvent(events []event, eventType string) *event { + for index := len(events) - 1; index >= 0; index-- { + if events[index].Type == eventType { + return &events[index] + } + } + return nil +} + +func lastReport(events []event, agentName string) *event { + for index := len(events) - 1; index >= 0; index-- { + if events[index].Type == "agent.thread_message_received" && + events[index].FromAgentName == agentName { + return &events[index] + } + } + return nil +} + +func (c *apiClient) events(ctx context.Context, sessionID string) ([]event, error) { + var history eventList + if err := c.get(ctx, "/v1/sessions/"+sessionID+"/events?order=asc&limit=1000", &history); err != nil { + return nil, err + } + return history.Data, nil +} + +func (c *apiClient) threads(ctx context.Context, sessionID string) ([]sessionThread, error) { + var output sessionThreadList + if err := c.get(ctx, "/v1/sessions/"+sessionID+"/threads?limit=1000", &output); err != nil { + return nil, err + } + return output.Data, nil +} + +func (c *apiClient) sendUserMessage(ctx context.Context, sessionID, text string) error { + return c.post(ctx, "/v1/sessions/"+sessionID+"/events", map[string]any{ + "events": []any{map[string]any{ + "type": "user.message", + "content": []any{map[string]any{ + "type": "text", "text": text, + }}, + }}, + }, nil) +} + +func (c *apiClient) create(ctx context.Context, endpoint string, input any) (resource, error) { + var output resource + if err := c.post(ctx, endpoint, input, &output); err != nil { + return resource{}, err + } + if output.ID == "" { + return resource{}, errors.New("response omitted id") + } + return output, nil +} + +func (c *apiClient) get(ctx context.Context, endpoint string, output any) error { + return c.do(ctx, http.MethodGet, endpoint, nil, output) +} + +func (c *apiClient) post(ctx context.Context, endpoint string, input, output any) error { + return c.do(ctx, http.MethodPost, endpoint, input, output) +} + +func (c *apiClient) do(ctx context.Context, method, endpoint string, input, output any) error { + var body io.Reader + if input != nil { + encoded, err := json.Marshal(input) + if err != nil { + return fmt.Errorf("encode request: %w", err) + } + body = bytes.NewReader(encoded) + } + req, err := http.NewRequestWithContext(ctx, method, c.baseURL+endpoint, body) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+c.apiKey) + if input != nil { + req.Header.Set("Content-Type", "application/json") + } + response, err := c.http.Do(req) + if err != nil { + return err + } + defer func() { _ = response.Body.Close() }() + data, err := io.ReadAll(io.LimitReader(response.Body, 1<<20)) + if err != nil { + return err + } + if response.StatusCode < 200 || response.StatusCode >= 300 { + return fmt.Errorf("%s %s returned %s: %s", method, endpoint, response.Status, strings.TrimSpace(string(data))) + } + if output == nil || len(bytes.TrimSpace(data)) == 0 { + return nil + } + if err := json.Unmarshal(data, output); err != nil { + return fmt.Errorf("decode %s %s response: %w", method, endpoint, err) + } + return nil +} + +func (c *apiClient) cleanup(ctx context.Context, resources createdResources) { + if resources.SessionID == "" && resources.EnvironmentID == "" && + len(resources.AgentIDs) == 0 { + return + } + fmt.Println("Cleaning up example resources...") + if resources.SessionID != "" { + if err := c.do(ctx, http.MethodDelete, "/v1/sessions/"+resources.SessionID, nil, nil); err != nil { + fmt.Fprintf(os.Stderr, "warning: cleanup Session failed: %v\n", err) + } + } + for index := len(resources.AgentIDs) - 1; index >= 0; index-- { + if err := c.do(ctx, http.MethodPost, "/v1/agents/"+resources.AgentIDs[index]+"/archive", nil, nil); err != nil { + fmt.Fprintf(os.Stderr, "warning: cleanup Agent %s failed: %v\n", resources.AgentIDs[index], err) + } + } + if resources.EnvironmentID != "" { + if err := c.do(ctx, http.MethodDelete, "/v1/environments/"+resources.EnvironmentID, nil, nil); err != nil { + fmt.Fprintf(os.Stderr, "warning: cleanup Environment failed: %v\n", err) + } + } +} diff --git a/internal/model/anthropic.go b/internal/model/anthropic.go index 688f7ad..e56bce6 100644 --- a/internal/model/anthropic.go +++ b/internal/model/anthropic.go @@ -80,15 +80,15 @@ func AnthropicFromEnv() (*Anthropic, bool, error) { // wireBlock is one content block in the Anthropic Messages wire format. A block // is a tagged union keyed on Type; only the fields relevant to that type are -// emitted (omitempty), so a text block carries text, a tool_use block carries -// id/name/input, and a tool_result block carries tool_use_id/content/is_error. +// emitted. Input is a pointer because Messages requires tool_use.input even for +// an empty object, while non-tool blocks must omit it. type wireBlock struct { Type string `json:"type"` Text string `json:"text,omitempty"` // tool_use - ID string `json:"id,omitempty"` - Name string `json:"name,omitempty"` - Input map[string]any `json:"input,omitempty"` + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Input *map[string]any `json:"input,omitempty"` // tool_result ToolUseID string `json:"tool_use_id,omitempty"` Content []json.RawMessage `json:"content,omitempty"` @@ -641,7 +641,13 @@ func marshalTypedBlock(b domain.ContentBlock) (json.RawMessage, error) { var value any switch b.Type { case "tool_use": - value = wireBlock{Type: "tool_use", ID: b.ToolUseID, Name: b.ToolName, Input: b.Input} + input := b.Input + if input == nil { + input = map[string]any{} + } + value = wireBlock{ + Type: "tool_use", ID: b.ToolUseID, Name: b.ToolName, Input: &input, + } case "tool_result": wb := wireBlock{Type: "tool_result", ToolUseID: b.ToolResultFor, IsError: b.IsError} if len(b.ResultContent) > 0 { diff --git a/internal/model/anthropic_test.go b/internal/model/anthropic_test.go index 051f99e..e2f8004 100644 --- a/internal/model/anthropic_test.go +++ b/internal/model/anthropic_test.go @@ -434,6 +434,27 @@ func TestAnthropic_SerializesToolsAndToolBlocks(t *testing.T) { } } +func TestAnthropic_SerializesEmptyToolUseInputAsObject(t *testing.T) { + for _, input := range []map[string]any{nil, {}} { + raw, err := marshalTypedBlock(domain.ContentBlock{ + Type: "tool_use", ToolUseID: "toolu_advisor", + ToolName: "advisor", Input: input, + }) + if err != nil { + t.Fatal(err) + } + var block map[string]any + if err := json.Unmarshal(raw, &block); err != nil { + t.Fatal(err) + } + value, present := block["input"] + object, objectOK := value.(map[string]any) + if !present || !objectOK || len(object) != 0 { + t.Fatalf("tool_use.input = %#v (present=%v), want required empty object; raw=%s", value, present, raw) + } + } +} + func TestAnthropic_SerializesRichToolResultContent(t *testing.T) { image := json.RawMessage(`{"type":"image","source":{"type":"url","url":"https://example.com/result.png"}}`) raw, err := marshalTypedBlock(domain.ContentBlock{ diff --git a/internal/pg/journal.go b/internal/pg/journal.go index bfc729d..ac37ce1 100644 --- a/internal/pg/journal.go +++ b/internal/pg/journal.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "reflect" + "time" "github.com/jackc/pgx/v5" "github.com/yanpgwang/mango/internal/domain" @@ -33,7 +34,12 @@ type TurnAttempt struct { // replayed. Call RecoverTurn first to classify leftovers. func (s *Store) BeginAttempt(ctx context.Context, sessionID, triggerEventID string) (TurnAttempt, error) { var attempt TurnAttempt - err := s.withTx(ctx, func(q *pgstore.Queries) error { + err := s.withPGXTx(ctx, func(tx pgx.Tx, q *pgstore.Queries) error { + if err := s.lockTurnExecutionOwner( + ctx, tx, q, sessionID, triggerEventID, + ); err != nil { + return err + } if _, err := q.ActiveAttemptForTurn(ctx, pgstore.ActiveAttemptForTurnParams{ SessionID: sessionID, TriggerEventID: triggerEventID, }); err == nil { @@ -80,6 +86,60 @@ func (s *Store) BeginAttempt(ctx context.Context, sessionID, triggerEventID stri return attempt, nil } +// lockTurnExecutionOwner serializes attempt admission with Session and Thread +// lifecycle transitions. Once an owner is terminal, no delayed Activity may +// create an attempt that a completed termination transaction could not fence. +func (s *Store) lockTurnExecutionOwner( + ctx context.Context, + tx pgx.Tx, + q *pgstore.Queries, + sessionID string, + triggerEventID string, +) error { + row, err := q.LockSession(ctx, sessionID) + if errors.Is(err, pgx.ErrNoRows) { + return domain.NotFound("session not found") + } + if err != nil { + return err + } + if row.DeletingAt.Valid { + return domain.Conflict("session deletion is in progress") + } + session, err := sessionFromLockRow(row) + if err != nil { + return err + } + if session.ArchivedAt != nil || session.Status == domain.StatusTerminated { + return domain.Conflict("cannot execute a turn for a terminated Session") + } + + var ( + threadStatus string + archivedAt *time.Time + ) + err = tx.QueryRow(ctx, ` +SELECT thread.status, thread.archived_at +FROM events AS event +JOIN session_threads AS thread + ON thread.session_id = event.session_id + AND thread.id = event.thread_id +WHERE event.session_id = $1 AND event.id = $2 +FOR UPDATE OF thread`, sessionID, triggerEventID).Scan( + &threadStatus, &archivedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return domain.NotFound("turn trigger event not found") + } + if err != nil { + return err + } + if archivedAt != nil || domain.Status(threadStatus) == domain.StatusTerminated { + return domain.Conflict("cannot execute a turn for a terminated Session Thread") + } + return nil +} + // EnsureAttempt returns the explicitly named active attempt for a // Workflow-owned turn, creating it when none exists. Unlike BeginAttempt, this // operation is intentionally idempotent: a Temporal Activity retry after the @@ -99,7 +159,12 @@ func (s *Store) EnsureAttempt( return TurnAttempt{}, domain.Validation("turn attempt id is required") } var attempt TurnAttempt - err := s.withTx(ctx, func(q *pgstore.Queries) error { + err := s.withPGXTx(ctx, func(tx pgx.Tx, q *pgstore.Queries) error { + if err := s.lockTurnExecutionOwner( + ctx, tx, q, sessionID, triggerEventID, + ); err != nil { + return err + } activeID, err := q.ActiveAttemptForTurn(ctx, pgstore.ActiveAttemptForTurnParams{ SessionID: sessionID, TriggerEventID: triggerEventID, }) @@ -217,11 +282,12 @@ func (s *Store) finishAttemptLocked( } now := s.clock.Now().UTC() if state == domain.RunAttemptInterrupted { - // WaitForCancellation guarantees the Activity has acknowledged the - // Workflow's cancellation before this transaction runs. Locking the - // attempt serializes this fence with StartToolStep: either Start won and - // its result-less step becomes ambiguous, or this interrupt wins and the - // stale prepared step can never cross the side-effect boundary. + // An ordinary interrupt reaches this point after the Activity acknowledges + // cancellation. Owner termination can also call it while committing the + // terminal lifecycle. Locking the attempt serializes both fences with + // StartToolStep: either Start already crossed the boundary and its + // result-less step becomes ambiguous, or this transition wins and a stale + // prepared step can never start. if _, err := q.MarkStartedStepsAmbiguousForAttempt( ctx, pgstore.MarkStartedStepsAmbiguousForAttemptParams{ diff --git a/internal/pg/multiagent_lifecycle_test.go b/internal/pg/multiagent_lifecycle_test.go index 53572d1..543a412 100644 --- a/internal/pg/multiagent_lifecycle_test.go +++ b/internal/pg/multiagent_lifecycle_test.go @@ -41,6 +41,20 @@ func TestArchiveSessionThread_RejectsRunningChild(t *testing.T) { func TestArchiveSessionThread_CommitsLifecycleAndTerminateIntentAtomically(t *testing.T) { fixture := newMultiagentInterruptFixture(t, "archive_idle") completeLifecycleChildIdle(t, fixture) + const staleAttemptID = "ratm_archive_idle_stale" + if _, err := fixture.store.EnsureAttempt( + fixture.ctx, fixture.session.ID, fixture.childTrigger.ID, staleAttemptID, + ); err != nil { + t.Fatal(err) + } + staleStep, err := fixture.store.EnsureToolStep( + fixture.ctx, staleAttemptID, "tstep_archive_idle_stale", 0, + "sevt_archive_idle_stale", "bash", + map[string]any{"command": "must not run after archive"}, + ) + if err != nil { + t.Fatal(err) + } archived, err := fixture.store.ArchiveSessionThread( fixture.ctx, fixture.session.ID, fixture.child.ID, @@ -52,6 +66,26 @@ func TestArchiveSessionThread_CommitsLifecycleAndTerminateIntentAtomically(t *te archived.Status != domain.StatusTerminated { t.Fatalf("archived child = %+v", archived) } + var staleAttemptState string + if err := fixture.store.pool.QueryRow( + fixture.ctx, `SELECT state FROM turn_attempts WHERE id = $1`, staleAttemptID, + ).Scan(&staleAttemptState); err != nil { + t.Fatal(err) + } + if staleAttemptState != string(domain.RunAttemptInterrupted) { + t.Fatalf("archived child attempt state = %s", staleAttemptState) + } + if err := fixture.store.StartToolStep(fixture.ctx, staleStep.ID); err == nil { + t.Fatal("prepared child step started after archive") + } + _, err = fixture.store.EnsureAttempt( + fixture.ctx, fixture.session.ID, fixture.childTrigger.ID, + "ratm_archive_idle_after_terminal", + ) + var domainErr *domain.DomainError + if !errors.As(err, &domainErr) || domainErr.Kind != domain.KindConflict { + t.Fatalf("attempt after child archive error = %v", err) + } childEvents, err := fixture.store.ThreadEventsAfter( fixture.ctx, fixture.session.ID, fixture.child.ID, 0, 100, @@ -248,6 +282,163 @@ WHERE session_id = $1 AND thread_id = $2`, } } +func TestPrimaryTerminalCompletion_FencesActiveChildrenAtomically(t *testing.T) { + fixture := newMultiagentInterruptFixture(t, "primary_terminal") + admitted, err := fixture.store.AdmitEvents( + fixture.ctx, fixture.session.ID, + []domain.EventDraft{{ + Type: domain.EvUserMessage, + Payload: map[string]any{"content": []any{map[string]any{ + "type": "text", "text": "stop after this failure", + }}}, + }}, + ) + if err != nil || len(admitted.SubmittedEvents) != 1 { + t.Fatalf("admit primary failure trigger = %+v, err=%v", admitted, err) + } + trigger := admitted.SubmittedEvents[0] + const attemptID = "ratm_primary_terminal" + if _, err := fixture.store.EnsureAttempt( + fixture.ctx, fixture.session.ID, trigger.ID, attemptID, + ); err != nil { + t.Fatal(err) + } + const childAttemptID = "ratm_child_fenced_by_primary" + if _, err := fixture.store.EnsureAttempt( + fixture.ctx, fixture.session.ID, fixture.childTrigger.ID, childAttemptID, + ); err != nil { + t.Fatal(err) + } + preparedStep, err := fixture.store.EnsureToolStep( + fixture.ctx, childAttemptID, "tstep_child_prepared_at_termination", 0, + "sevt_child_prepared_at_termination", "bash", + map[string]any{"command": "must not run"}, + ) + if err != nil { + t.Fatal(err) + } + startedStep, err := fixture.store.EnsureToolStep( + fixture.ctx, childAttemptID, "tstep_child_started_at_termination", 1, + "sevt_child_started_at_termination", "bash", + map[string]any{"command": "may already have run"}, + ) + if err != nil { + t.Fatal(err) + } + if err := fixture.store.StartToolStep(fixture.ctx, startedStep.ID); err != nil { + t.Fatal(err) + } + attemptError := "primary model request failed" + completion, err := fixture.store.CompleteWorkflowTurn( + fixture.ctx, fixture.session.ID, trigger.ID, + []domain.EventDraft{ + {Type: domain.EvSessionError, Payload: map[string]any{ + "error": map[string]any{ + "type": "model_request_failed_error", "message": attemptError, + "retry_status": map[string]any{"type": "terminal"}, + }, + }}, + {Type: domain.EvSessionStatusTerminated, Payload: map[string]any{}}, + }, + domain.StatusTerminated, attemptID, domain.RunAttemptFailed, + &attemptError, nil, nil, + ) + if err != nil || completion.Session.Status != domain.StatusTerminated { + t.Fatalf("primary terminal completion = %+v, err=%v", completion, err) + } + + child, err := fixture.store.GetSessionThread( + fixture.ctx, fixture.session.ID, fixture.child.ID, + ) + if err != nil || child.Status != domain.StatusTerminated { + t.Fatalf("fenced child = %+v, err=%v", child, err) + } + var childAttemptState, preparedState, startedState string + if err := fixture.store.pool.QueryRow(fixture.ctx, ` +SELECT attempt.state, prepared.state, started.state +FROM turn_attempts AS attempt +JOIN tool_steps AS prepared ON prepared.id = $2 +JOIN tool_steps AS started ON started.id = $3 +WHERE attempt.id = $1`, childAttemptID, preparedStep.ID, startedStep.ID).Scan( + &childAttemptState, &preparedState, &startedState, + ); err != nil { + t.Fatal(err) + } + if childAttemptState != string(domain.RunAttemptInterrupted) || + preparedState != string(domain.ToolStepPrepared) || + startedState != string(domain.ToolStepAmbiguous) { + t.Fatalf( + "fenced journal attempt=%s prepared=%s started=%s", + childAttemptState, preparedState, startedState, + ) + } + if err := fixture.store.StartToolStep(fixture.ctx, preparedStep.ID); err == nil { + t.Fatal("prepared child step started after primary termination") + } + _, err = fixture.store.EnsureAttempt( + fixture.ctx, fixture.session.ID, fixture.childTrigger.ID, + "ratm_child_created_after_primary_termination", + ) + var domainErr *domain.DomainError + if !errors.As(err, &domainErr) || domainErr.Kind != domain.KindConflict { + t.Fatalf("late child attempt error = %v", err) + } + childEvents, err := fixture.store.ThreadEventsAfter( + fixture.ctx, fixture.session.ID, fixture.child.ID, 0, 100, + ) + if err != nil || countEventsOfType( + childEvents, domain.EvSessionThreadStatusTerminated, + ) != 1 { + t.Fatalf("child termination events = %+v, err=%v", childEvents, err) + } + primaryEvents, err := fixture.store.QueryEvents( + fixture.ctx, fixture.session.ID, app.EventQuery{Limit: 100}, + ) + if err != nil || countEventsOfType( + primaryEvents, domain.EvSessionThreadStatusTerminated, + ) != 1 { + t.Fatalf("primary termination projection = %+v, err=%v", primaryEvents, err) + } + + var intent string + if err := fixture.store.pool.QueryRow(fixture.ctx, ` +SELECT intent +FROM thread_orchestration_outbox +WHERE session_id = $1 AND thread_id = $2`, + fixture.session.ID, fixture.child.ID, + ).Scan(&intent); err != nil { + t.Fatal(err) + } + if intent != string(OrchestrationTerminate) { + t.Fatalf("child termination intent = %q", intent) + } + + late, err := fixture.store.CompleteThreadWorkflowTurn( + fixture.ctx, fixture.session.ID, fixture.child.ID, + fixture.childTrigger.ID, + []domain.EventDraft{{ + Type: domain.EvSessionStatusIdle, + Payload: map[string]any{ + "stop_reason": map[string]any{"type": "end_turn"}, + }, + }}, + domain.StatusIdle, "", "", nil, nil, nil, nil, nil, + domain.TokenUsage{}, + ) + if err != nil || late.Applied || late.ThreadStatus != domain.StatusTerminated { + t.Fatalf("late child completion = %+v, err=%v", late, err) + } + primaryEvents, err = fixture.store.QueryEvents( + fixture.ctx, fixture.session.ID, app.EventQuery{Limit: 100}, + ) + if err != nil { + t.Fatal(err) + } + if countEventsOfType(primaryEvents, domain.EvAgentThreadMessageReceived) != 0 { + t.Fatalf("late child report reached terminated primary: %+v", primaryEvents) + } +} + func countEventsOfType(events []domain.Event, eventType string) int { count := 0 for _, event := range events { diff --git a/internal/pg/session_threads.go b/internal/pg/session_threads.go index 711cb7c..70965a2 100644 --- a/internal/pg/session_threads.go +++ b/internal/pg/session_threads.go @@ -221,6 +221,11 @@ WHERE session_id = $1 AND id = $2`, sessionID, threadID).Scan(&threadKind); err "cannot archive a running session Thread; interrupt first", ) } + if err := s.interruptSessionThreadAttemptsLocked( + ctx, tx, q, sessionID, threadID, + ); err != nil { + return err + } pendingResult, err := tx.Exec(ctx, ` DELETE FROM pending_actions @@ -386,6 +391,165 @@ ORDER BY created_at, id`, sessionID) return nil } +// interruptSessionThreadAttemptsLocked closes every active execution journal +// owned by one Thread. The caller holds the Session and Thread lifecycle locks; +// locking each attempt establishes the final ordering with StartToolStep. +func (s *Store) interruptSessionThreadAttemptsLocked( + ctx context.Context, + tx pgx.Tx, + q *pgstore.Queries, + sessionID string, + threadID string, +) error { + rows, err := tx.Query(ctx, ` +SELECT attempt.id, attempt.trigger_event_id +FROM turn_attempts AS attempt +JOIN events AS trigger + ON trigger.session_id = attempt.session_id + AND trigger.id = attempt.trigger_event_id +WHERE attempt.session_id = $1 + AND trigger.thread_id = $2 + AND attempt.state = 'active' +ORDER BY attempt.id +FOR UPDATE OF attempt`, sessionID, threadID) + if err != nil { + return err + } + type activeAttempt struct { + id string + triggerEventID string + } + attempts := make([]activeAttempt, 0) + for rows.Next() { + var attempt activeAttempt + if err := rows.Scan(&attempt.id, &attempt.triggerEventID); err != nil { + rows.Close() + return err + } + attempts = append(attempts, attempt) + } + if err := rows.Err(); err != nil { + rows.Close() + return err + } + rows.Close() + + for _, attempt := range attempts { + if err := s.finishAttemptLocked( + ctx, q, attempt.id, domain.RunAttemptInterrupted, nil, + sessionID, attempt.triggerEventID, + ); err != nil { + return err + } + } + return nil +} + +// terminateChildSessionThreadsLocked fences every persistent child when its +// primary owner reaches a terminal state. Child completion takes the same +// Session row lock, so either it commits before this transition or observes the +// terminated projection and becomes a no-op. Lifecycle events and durable +// Workflow termination intents commit in this same transaction. +func (s *Store) terminateChildSessionThreadsLocked( + ctx context.Context, + tx pgx.Tx, + q *pgstore.Queries, + sessionID string, + primaryThreadID string, + triggerEventID string, + maxSeq int64, + now time.Time, +) ([]domain.Event, int64, error) { + rows, err := tx.Query(ctx, ` +SELECT body +FROM session_threads +WHERE session_id = $1 + AND kind = 'child' + AND archived_at IS NULL + AND status <> 'terminated' +ORDER BY created_at, id +FOR UPDATE`, sessionID) + if err != nil { + return nil, maxSeq, err + } + children := make([]domain.SessionThread, 0) + for rows.Next() { + var body []byte + if err := rows.Scan(&body); err != nil { + rows.Close() + return nil, maxSeq, err + } + var thread domain.SessionThread + if err := json.Unmarshal(body, &thread); err != nil { + rows.Close() + return nil, maxSeq, err + } + children = append(children, thread) + } + if err := rows.Err(); err != nil { + rows.Close() + return nil, maxSeq, err + } + rows.Close() + + committed := make([]domain.Event, 0, len(children)*2) + for _, thread := range children { + if err := s.interruptSessionThreadAttemptsLocked( + ctx, tx, q, sessionID, thread.ID, + ); err != nil { + return nil, maxSeq, err + } + if _, err := tx.Exec(ctx, ` +DELETE FROM pending_actions +WHERE session_id = $1 AND thread_id = $2 AND resolved_at IS NULL`, + sessionID, thread.ID, + ); err != nil { + return nil, maxSeq, err + } + if _, err := tx.Exec(ctx, ` +UPDATE events +SET processed_at = COALESCE(processed_at, $3) +WHERE session_id = $1 AND thread_id = $2 AND processed_at IS NULL`, + sessionID, thread.ID, now, + ); err != nil { + return nil, maxSeq, err + } + + thread.TransitionStatus(domain.StatusTerminated, now) + if err := putSessionThreadTx(ctx, tx, thread); err != nil { + return nil, maxSeq, err + } + lifecycle := domain.EventDraft{ + Type: domain.EvSessionThreadStatusTerminated, + Payload: threadLifecyclePayload(thread), + } + childEvents, next, err := s.appendThreadDraftsAt( + ctx, q, sessionID, thread.ID, + []domain.EventDraft{lifecycle}, maxSeq, nil, now, + ) + if err != nil { + return nil, maxSeq, err + } + maxSeq = next + committed = append(committed, childEvents...) + primaryEvents, next, err := s.appendThreadDraftsAt( + ctx, q, sessionID, primaryThreadID, + []domain.EventDraft{lifecycle}, maxSeq, &triggerEventID, now, + ) + if err != nil { + return nil, maxSeq, err + } + maxSeq = next + committed = append(committed, primaryEvents...) + } + if err := enqueueChildThreadTerminationsLocked( + ctx, tx, q, sessionID, maxSeq, now, + ); err != nil { + return nil, maxSeq, err + } + return committed, maxSeq, nil +} + // CreateChildSessionThread atomically captures a callable Agent from the // Session-owned resolved roster, inserts its independent Thread projection, // and appends session.thread_created to the parent ledger. It deliberately does diff --git a/internal/pg/store.go b/internal/pg/store.go index 33b22a0..0c03a6b 100644 --- a/internal/pg/store.go +++ b/internal/pg/store.go @@ -2061,7 +2061,7 @@ func (s *Store) completeTurn( usage domain.TokenUsage, ) (TurnCompletion, error) { var result TurnCompletion - err := s.withTx(ctx, func(q *pgstore.Queries) error { + err := s.withPGXTx(ctx, func(tx pgx.Tx, q *pgstore.Queries) error { row, err := q.LockSession(ctx, sessionID) if errors.Is(err, pgx.ErrNoRows) { return domain.NotFound("session not found") @@ -2383,6 +2383,17 @@ func (s *Store) completeTurn( if err != nil { return err } + if independentPrimary && effectiveStatus == domain.StatusTerminated { + terminatedEvents, next, err := s.terminateChildSessionThreadsLocked( + ctx, tx, q, sessionID, primaryThread.ID, + triggerEventID, finalMaxSeq, completionTime, + ) + if err != nil { + return err + } + events = append(events, terminatedEvents...) + finalMaxSeq = next + } allowedActions := make(map[string]domain.Event, len(events)) for _, event := range events { allowedActions[event.ID] = event diff --git a/internal/pg/thread_completion.go b/internal/pg/thread_completion.go index f640e2c..5f21799 100644 --- a/internal/pg/thread_completion.go +++ b/internal/pg/thread_completion.go @@ -169,6 +169,13 @@ func (s *Store) CompleteThreadWorkflowTurn( return err } } + if status == domain.StatusTerminated { + if err := s.interruptSessionThreadAttemptsLocked( + ctx, tx, q, sessionID, threadID, + ); err != nil { + return err + } + } resolvedPending, err := s.resolvePendingBarrierLocked( ctx, q, sessionID, threadID, triggerEventID, resolutionEventIDs, ) diff --git a/website/sidebars.ts b/website/sidebars.ts index 07945ce..0763010 100644 --- a/website/sidebars.ts +++ b/website/sidebars.ts @@ -43,6 +43,7 @@ const sidebars: SidebarsConfig = { items: [ 'examples/coding-agent-iterate', 'examples/hitl-gate', + 'examples/multi-agent-team', ], }, {