diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 3bcc4d162..2a1114e0c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -59,6 +59,34 @@ isolation remain separate work. the pinned public API, or authorize unrelated refactors or live Session configuration switching. +## Core–Runtime integration contract + +The [Core–Runtime protocol](docs/runtime-protocol.md) is the integration entry +point for message order, identities, receipts, capability declarations and failure +ownership. The shared types and validators live only in +`internal/agentdaemon/proto`; do not introduce a parallel schema or per-Harness +orchestration path. Both hosted and self-hosted peers use the same contract. +Protocol changes update matching peers together, with an exact wire-version +check and no historical wire fallback. + +Core queues and Runtime writes are transport observations, not native acceptance +or completion. The gateway exposes only durable Run subscriptions: connection +loss or overflow closes the subscription with an error and never fabricates +native error/done events. A timeout leaves unconfirmed effects unknown; transport +reconnection never authorizes mutation replay. The public Turn may still be +`failed` for an observation failure (`delivery_unknown` or +`event_stream_incomplete`); that status does not prove native failure or cleanup. +Runtime owns native settlement +and cleanup; Executor release does not reclaim Sandbox Provider compute. +The daemon runs with its starting user's authority, not as an isolation boundary; +Core-managed sandbox infrastructure owns outer isolation. + +`make check-runtime-contract` is the focused shared-contract entry point. Its +wire, gateway, transport, dispatcher, real WebSocket and observation-result tests +also run in the required `make check` through `check-go` and `check-agents-api`. Keep failure, cancellation, disconnect +and cleanup assertions in that common suite; native adapter acceptance remains +required for advertised capabilities. + ## Workflow and quality Develop in an isolated worktree on a feature branch and submit a PR. Do not edit @@ -496,7 +524,7 @@ forwarding. The explicit daemon-executor decision supersedes the previous native executor interoperability requirement. The superseded execution route is removed; retain reusable filesystem helpers, necessary regressions and historical evidence without a compatibility layer. -The private daemon wire protocol is 0.7.0. Initial, prepared and active input use +The private daemon wire protocol is 0.8.0. Initial, prepared and active input use the same ordered MessageInput contract, replacing scalar prompts and attachments. User-message boundaries and text/image order remain intact through Core and the Runtime wire; adapters own native conversion and receipt aggregation. Text-only @@ -506,8 +534,8 @@ unqualified harnesses (Claude SDK, MiniMax Code) reject them at admission rather their input rewritten. Codex has a flat native input list and uses blank-line separators between messages; this does not preserve independent native user-message boundaries. No old wire fallback is maintained. -Deploy Core and daemon together; the existing major/minor WebSocket check rejects -older major/minor peers before dispatch rather than ignoring removed fields. +Deploy Core and daemon together; the WebSocket check requires the exact wire +version, including patch, and rejects mismatches before dispatch. The independently packaged Claude bridge uses protocol 3 for a prepared Executor and separately identified Turns; readiness rejects other protocol versions. Image-bearing messages require a qualified profile/placement before persistence @@ -1176,7 +1204,7 @@ a chain of old writable disks across suspension cycles. No Kubernetes, distribut scheduler or snapshot replication belongs in this V1 profile. Queued work and live Environment file access request wake. History and published -artifact reads do not. Planned suspension uses private daemon wire 0.7.0 with an +artifact reads do not. Planned suspension uses private daemon wire 0.8.0 with an Environment and suspension token; a PID/start-time fenced local control signal wakes the parked daemon, which reauthenticates before admitting new work. A transient disconnect before confirmation retries the same armed suspension with @@ -1278,11 +1306,12 @@ through the provider-neutral bootstrap and checked against execution preparation The shared policy includes enabled, disabled and an exact-host restricted allowlist. Core preserves public spelling/order/duplicates and only permits Template overrides that narrow authority. Adapters translate a normalized copy into native settings; -Core and Docker never select native profile names. Every hosted execution peer -must support `local_environment_network_policy` and receive the complete bound -policy; missing policy never falls back to enabled or an older peer path. Read-only -workspace access does not require execution network policy. A declaration alone -does not qualify an image or admit public hosted creation. +Core and Docker never select native profile names. Execution preparation receives +the complete bound policy; missing policy never falls back to enabled or an older +peer path. Network isolation belongs to the surrounding sandbox, not a daemon +capability declaration. Read-only workspace access does not require execution +network policy. A declaration alone does not qualify an image or admit public +hosted creation. Codex restricted networking uses its native managed network requirements and proxy. Its adapter preserves the image's filesystem, approval and hook requirements and diff --git a/Makefile b/Makefile index d72d18bdd..052bf1756 100644 --- a/Makefile +++ b/Makefile @@ -44,6 +44,11 @@ check-sqlc: check-go: go test ./apps/parsar-daemon/... ./internal/... ./contracts/agents-api/... ./scripts/openapi-split -count=1 +.PHONY: check-runtime-contract +check-runtime-contract: + go test ./internal/agentdaemon/proto ./internal/agentdaemon/gateway ./apps/parsar-daemon/internal/transport ./apps/parsar-daemon/internal/dispatch ./apps/parsar-daemon/internal/contracttest -count=1 + go test ./services/agents-api/internal/execution -run '^TestRuntimeProtocol' -count=1 + build-daemon: @set -e; output="$${OAC_DEV_HOME:-$$HOME/.oac}/build/daemon"; \ [[ "$$output" == /* ]] || { echo 'Daemon output directory must be absolute' >&2; exit 1; }; \ diff --git a/README.md b/README.md index c8f255455..bfadf51e2 100644 --- a/README.md +++ b/README.md @@ -81,6 +81,7 @@ native differences. | [API reference](docs/api/README.md) | The `/v1`, `/core/v1` and `/api/v1` namespaces | | [Nodes and sandbox backends: operator reference](services/agents-api/HOSTED-SANDBOX-MANAGER.md) | Node protocol, manual registration, placement, reset | | [Protocol coverage](contracts/agents-api/README.md) and [harness selection](contracts/agents-api/harness-selection.md) | Supported operations and native differences | +| [Core–Runtime protocol](docs/runtime-protocol.md) | Runtime integration, messages, receipts, failure ownership and contract checks | | [Add a Harness](contracts/agents-api/harness-onboarding.md) | Required adapter interfaces, lifecycle, capabilities, registration and acceptance | | [Landing page source](site/index.html) | The public site | diff --git a/apps/parsar-daemon/internal/cli/agent_discovery.go b/apps/parsar-daemon/internal/cli/agent_discovery.go index ad727c627..a2233066a 100644 --- a/apps/parsar-daemon/internal/cli/agent_discovery.go +++ b/apps/parsar-daemon/internal/cli/agent_discovery.go @@ -155,7 +155,6 @@ func discoverAgentCLIs(rc *runContext, profile string, checks agentCLIChecks) (a out.Codex.Version = codexVersion out.Codex.Capabilities.NativeSessionRecovery = codex.SupportsNativeSessionRecovery(codexVersion) out.Codex.Capabilities.LocalEnvironment = codex.SupportsLocalEnvironment(codexVersion) - out.Codex.Capabilities.LocalEnvironmentNetworkPolicy = codex.SupportsLocalNetworkPolicy(codexVersion) out.Codex.Capabilities.MCPHTTPRequired = codex.SupportsNativeSessionRecovery(codexVersion) fmt.Fprintf(rc.stdout, "Codex preflight ok (%s)\n", codexVersion) } else if errors.Is(codexErr, codex.ErrCLINotFound) { diff --git a/apps/parsar-daemon/internal/cli/claude_sdk.go b/apps/parsar-daemon/internal/cli/claude_sdk.go index 746288935..f0b3ea151 100644 --- a/apps/parsar-daemon/internal/cli/claude_sdk.go +++ b/apps/parsar-daemon/internal/cli/claude_sdk.go @@ -94,7 +94,7 @@ func discoverClaudeSDK(rc *runContext, profile string, check func(context.Contex } caps := &out.Info.Capabilities caps.EnvironmentNone, caps.FunctionTools = false, info.SupportsWorkspaceFunctions() - caps.Preparation, caps.LocalEnvironment, caps.LocalEnvironmentNetworkPolicy = true, true, true + caps.Preparation, caps.LocalEnvironment = true, true caps.WorkspaceReadPreparation, caps.NativeSessionRecovery = true, true } out.Info.Available, out.Info.Version = true, info.SDK diff --git a/apps/parsar-daemon/internal/cli/connect.go b/apps/parsar-daemon/internal/cli/connect.go index 61f0ebe4a..e76aacb35 100644 --- a/apps/parsar-daemon/internal/cli/connect.go +++ b/apps/parsar-daemon/internal/cli/connect.go @@ -309,7 +309,7 @@ func mainLoopRemote(rc *runContext, profile string, prof auth.Profile, agentCLIs DeviceID: boot.DeviceID, Credential: prof.RunnerCredential, // DaemonVersion is the WIRE-PROTOCOL version, not the build - // tag. proto.VersionCompatible is a strict major.minor + // tag. proto.VersionCompatible requires an exact version // match against proto.Version. Build-tag reporting goes // in heartbeat's DaemonVersion field. DaemonVersion: proto.Version, @@ -420,7 +420,6 @@ func pumpConn(parentCtx context.Context, conn *transport.Conn, registry *agent.R Timestamp: time.Now().Unix(), ActiveRequests: router.ActiveRuns(), DaemonVersion: Version, - ClaudeAvailable: agentCLIs.ClaudeCode.Available, // legacy server compatibility SupportedAgentKinds: kinds, } }, obslog.Bg().With("component", "heartbeat")) diff --git a/apps/parsar-daemon/internal/cli/connect_suspend.go b/apps/parsar-daemon/internal/cli/connect_suspend.go index c471a12f0..0253f0a5f 100644 --- a/apps/parsar-daemon/internal/cli/connect_suspend.go +++ b/apps/parsar-daemon/internal/cli/connect_suspend.go @@ -161,7 +161,7 @@ func (s *suspendedRouter) heartbeats(ctx context.Context, conn *transport.Conn, caps := &kinds[i].Capabilities caps.WorkspaceOutputExport = s.local.CanExport() && caps.LocalEnvironment && caps.WorkspaceReadPreparation } - return proto.HeartbeatPayload{Timestamp: time.Now().Unix(), ActiveRequests: s.router.ActiveRuns(), DaemonVersion: Version, ClaudeAvailable: discovery.ClaudeCode.Available, SupportedAgentKinds: kinds} + return proto.HeartbeatPayload{Timestamp: time.Now().Unix(), ActiveRequests: s.router.ActiveRuns(), DaemonVersion: Version, SupportedAgentKinds: kinds} }, obslog.Bg()) } diff --git a/apps/parsar-daemon/internal/cli/mcode_workspace.go b/apps/parsar-daemon/internal/cli/mcode_workspace.go index 9e7d45f8d..9d5f37922 100644 --- a/apps/parsar-daemon/internal/cli/mcode_workspace.go +++ b/apps/parsar-daemon/internal/cli/mcode_workspace.go @@ -71,6 +71,6 @@ func discoverMCodeWorkspace(rc *runContext, discovery *agentCLIDiscovery) { discovery.MCodeWorkspace = &c caps := &discovery.MCode.Capabilities caps.EnvironmentNone = false - caps.Preparation, caps.LocalEnvironment, caps.LocalEnvironmentNetworkPolicy = true, true, true + caps.Preparation, caps.LocalEnvironment = true, true caps.WorkspaceReadPreparation = true } diff --git a/apps/parsar-daemon/internal/contracttest/wire_test.go b/apps/parsar-daemon/internal/contracttest/wire_test.go new file mode 100644 index 000000000..d2b38ca98 --- /dev/null +++ b/apps/parsar-daemon/internal/contracttest/wire_test.go @@ -0,0 +1,374 @@ +// Package contracttest exercises the shared protocol over an actual WebSocket, +// using the production gateway, transport and dispatcher with a controlled adapter. +package contracttest + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/MiniMax-AI-Dev/parsar/apps/parsar-daemon/internal/agent" + "github.com/MiniMax-AI-Dev/parsar/apps/parsar-daemon/internal/dispatch" + "github.com/MiniMax-AI-Dev/parsar/apps/parsar-daemon/internal/transport" + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/device" + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/gateway" + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" +) + +const credential = "synthetic-contract-credential" + +type credentialStore struct{} + +func (credentialStore) GetDeviceCredential(context.Context, string) (device.Credential, bool, error) { + return device.Credential{ID: "runtime", WorkspaceID: "tenant", Type: gateway.RuntimeTypeAgentDaemon, CredentialHash: device.HashCredential(credential)}, true, nil +} + +func newGateway(t *testing.T) (*gateway.Registry, string) { + t.Helper() + reg := gateway.NewRegistry() + handler := gateway.NewHandler(gateway.HandlerConfig{Registry: reg, Authenticator: gateway.NewAuthenticator(credentialStore{})}) + server := httptest.NewServer(http.HandlerFunc(handler.WS)) + t.Cleanup(server.Close) + return reg, "ws" + strings.TrimPrefix(server.URL, "http") +} + +func dial(t *testing.T, endpoint, version string) (*transport.Conn, error) { + t.Helper() + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + defer cancel() + return transport.Dial(ctx, transport.DialOptions{WSURL: endpoint, DeviceID: "runtime", Credential: credential, DaemonVersion: version}) +} + +func TestWireContractRejectsVersionMismatchWithoutReconnect(t *testing.T) { + _, endpoint := newGateway(t) + for _, version := range []string{"0.7.0", "0.8.99", "0.8.", proto.Version + "-dev", proto.Version + "+build"} { + t.Run(version, func(t *testing.T) { + attempts := 0 + _, err := transport.Reconnect(t.Context(), func(context.Context) (*transport.Conn, error) { + attempts++ + return dial(t, endpoint, version) + }, transport.DefaultBackoff, nil) + if !errors.Is(err, transport.ErrIncompatibleVersion) || !errors.Is(err, transport.ErrPermanent) || attempts != 1 { + t.Fatalf("mismatch must stop after one attempt: attempts=%d error=%v", attempts, err) + } + }) + } +} + +type wireFixture struct { + core *gateway.Session + conn *transport.Conn + router *dispatch.Router + reg *gateway.Registry + endpoint string + stopped chan struct{} + shutdownErr error +} + +func connectFixture(t *testing.T, factory agent.ExecutorFactory) *wireFixture { + t.Helper() + reg, endpoint := newGateway(t) + conn, err := dial(t, endpoint, proto.Version) + if err != nil { + t.Fatal(err) + } + core, err := reg.WaitForDevice(t.Context(), "runtime", 3*time.Second) + if err != nil { + t.Fatal(err) + } + kinds := agent.NewRegistry() + kinds.RegisterKind(proto.SupportedAgentKind{Kind: "contract", Available: true, Capabilities: proto.AgentKindCapabilities{EnvironmentNone: true}}, + func(context.Context, proto.PromptRequestPayload, chan<- proto.Envelope) (agent.Session, error) { + return nil, errors.New("prepared execution must not use prompt_request") + }) + kinds.RegisterExecutor("contract", factory) + router, err := dispatch.New(dispatch.Config{Registry: kinds, Sender: conn}) + if err != nil { + t.Fatal(err) + } + f := &wireFixture{core: core, conn: conn, router: router, reg: reg, endpoint: endpoint, stopped: make(chan struct{})} + go func() { + for env := range conn.Recv() { + if err := router.Handle(t.Context(), env); err != nil { + t.Errorf("dispatch: %v", err) + } + } + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + f.shutdownErr = router.Shutdown(ctx) + close(f.stopped) + }() + t.Cleanup(func() { + conn.Close() + select { + case <-f.stopped: + if f.shutdownErr != nil { + t.Error(f.shutdownErr) + } + case <-time.After(4 * time.Second): + t.Error("Runtime shutdown did not complete") + } + core.Close("test finished") + }) + return f +} + +func (f *wireFixture) send(t *testing.T, kind, id string, payload any) { + t.Helper() + env, err := proto.NewEnvelope(kind, id, payload) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + defer cancel() + if err := f.core.Send(ctx, env); err != nil { + t.Fatal(err) + } +} + +func receive(t *testing.T, sub *gateway.Subscription) proto.Envelope { + t.Helper() + select { + case env, ok := <-sub.Events: + if !ok { + t.Fatalf("subscription closed: %v", sub.Err()) + } + return env + case <-time.After(3 * time.Second): + t.Fatal("no protocol response") + } + return proto.Envelope{} +} + +func status(t *testing.T, sub *gateway.Subscription, want string) proto.PreparationStatusPayload { + t.Helper() + for { + env := receive(t, sub) + var got proto.PreparationStatusPayload + if env.Type != proto.TypePreparationStatus || env.DecodePayload(&got) != nil { + t.Fatalf("invalid preparation response: %+v", env) + } + if got.State == want { + return got + } + if got.State != "preparing" && got.State != "starting" { + t.Fatalf("preparation state=%s, want=%s", got.State, want) + } + } +} + +func (f *wireFixture) prepare(t *testing.T) (*gateway.Subscription, proto.PreparationStatusPayload) { + t.Helper() + sub, err := f.core.SubscribePreparation("prepare") + if err != nil { + t.Fatal(err) + } + f.send(t, proto.TypeExecutionPrepare, "prepare", proto.ExecutionPreparePayload{ + SessionID: "session", Configuration: proto.PromptRequestPayload{AgentKind: "contract", AgentStateKey: "agents-api-session", StrictResume: true, DisableExecutionEnvironment: true}, + }) + return sub, status(t, sub, "ready") +} + +type controlledExecutor struct { + turn chan *controlledTurn + starts atomic.Int32 + closes atomic.Int32 +} + +func newExecutor() *controlledExecutor { + return &controlledExecutor{turn: make(chan *controlledTurn, 1)} +} +func (e *controlledExecutor) Close(context.Context) error { e.closes.Add(1); return nil } +func (e *controlledExecutor) StartTurn(_ context.Context, id string, _ proto.MessageInput, out chan<- proto.Envelope) (agent.Turn, error) { + e.starts.Add(1) + turn := &controlledTurn{id: id, out: out, cancelling: make(chan struct{}), allowCancel: make(chan struct{}), settled: make(chan struct{})} + e.turn <- turn + return turn, nil +} + +type controlledTurn struct { + id string + out chan<- proto.Envelope + cancelling, allowCancel, settled chan struct{} + cancelOnce, finishOnce sync.Once +} + +func (turn *controlledTurn) Cancel(ctx context.Context) error { + turn.cancelOnce.Do(func() { close(turn.cancelling) }) + select { + case <-turn.allowCancel: + turn.finishOnce.Do(func() { close(turn.out); close(turn.settled) }) + return nil + case <-ctx.Done(): + return ctx.Err() + } +} +func (turn *controlledTurn) CancellationOutcome() proto.DonePayload { + return proto.DonePayload{Content: "partial", Metadata: map[string]any{proto.DoneMetaAgentSessionID: "native-session"}} +} +func (turn *controlledTurn) AwaitSettlement(ctx context.Context) (agent.TurnSettlement, error) { + select { + case <-turn.settled: + return agent.TurnSettlement{Reason: "cancelled fixture"}, nil + case <-ctx.Done(): + return agent.TurnSettlement{}, ctx.Err() + } +} + +func TestWireContractCancellationWaitsForSettlement(t *testing.T) { + executor := newExecutor() + f := connectFixture(t, func(context.Context, proto.PromptRequestPayload) (agent.Executor, error) { return executor, nil }) + prepared, ready := f.prepare(t) + if executor.starts.Load() != 0 { + t.Fatal("preparation submitted input") + } + run, err := f.core.SubscribeDurable("run") + if err != nil { + t.Fatal(err) + } + f.send(t, proto.TypeExecutionStart, "prepare", proto.ExecutionStartPayload{ExecutorID: ready.ExecutorID, Handle: ready.Handle, RunID: "run", Input: proto.TextInput("hello")}) + started := status(t, prepared, "started") + if started.RunID != "run" || started.ExecutorID != ready.ExecutorID { + t.Fatal("start lost execution identity") + } + var turn *controlledTurn + select { + case turn = <-executor.turn: + case <-time.After(3 * time.Second): + t.Fatal("no native turn") + } + var unblock sync.Once + release := func() { unblock.Do(func() { close(turn.allowCancel) }) } + t.Cleanup(release) + result := make(chan proto.InteractionDecisionAckPayload, 1) + ackErr := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(t.Context(), 3*time.Second) + defer cancel() + env, _ := proto.NewEnvelope(proto.TypePromptCancel, "run", proto.PromptCancelPayload{DeliveryID: "cancel"}) + ack, err := f.core.SendAndWaitInteractionAck(ctx, env, "cancel") + result <- ack + ackErr <- err + }() + select { + case <-turn.cancelling: + case <-time.After(3 * time.Second): + t.Fatal("cancellation was not received") + } + select { + case <-result: + t.Fatal("receipt preceded native settlement") + default: + } + release() + select { + case ack := <-result: + if err := <-ackErr; err != nil || !ack.Applied || ack.Outcome == nil || ack.Outcome.Content != "partial" { + t.Fatalf("invalid cancellation receipt: %+v, %v", ack, err) + } + case <-time.After(3 * time.Second): + t.Fatal("missing settled cancellation receipt") + } + f.core.Unsubscribe("run") + for env := range run.Events { + if env.ID != "run" { + t.Fatal("settled cancellation output lost Run correlation") + } + } + if executor.starts.Load() != 1 { + t.Fatal("input was submitted more than once") + } +} + +func TestWireContractPreparationFailureCleansUpWithoutRunCompletion(t *testing.T) { + executor := newExecutor() + f := connectFixture(t, func(context.Context, proto.PromptRequestPayload) (agent.Executor, error) { + return executor, errors.New("native setup failed") + }) + sub, err := f.core.SubscribePreparation("prepare") + if err != nil { + t.Fatal(err) + } + run, err := f.core.SubscribeDurable("run") + if err != nil { + t.Fatal(err) + } + f.send(t, proto.TypeExecutionPrepare, "prepare", proto.ExecutionPreparePayload{SessionID: "session", Configuration: proto.PromptRequestPayload{AgentKind: "contract", AgentStateKey: "agents-api-session", StrictResume: true, DisableExecutionEnvironment: true}}) + failed := status(t, sub, "failed") + if failed.ErrorCode != "preparation_failed" || executor.starts.Load() != 0 || executor.closes.Load() != 1 { + t.Fatalf("failed preparation retained resources or started input: %+v starts=%d closes=%d", failed, executor.starts.Load(), executor.closes.Load()) + } + if len(run.Events) != 0 { + t.Fatal("preparation failure manufactured a Run result") + } +} + +func TestWireContractDisconnectIsUnknownAndReconnectDoesNotReplay(t *testing.T) { + executor := newExecutor() + f := connectFixture(t, func(context.Context, proto.PromptRequestPayload) (agent.Executor, error) { return executor, nil }) + prepared, ready := f.prepare(t) + run, err := f.core.SubscribeDurable("run") + if err != nil { + t.Fatal(err) + } + f.send(t, proto.TypeExecutionStart, "prepare", proto.ExecutionStartPayload{ExecutorID: ready.ExecutorID, Handle: ready.Handle, RunID: "run", Input: proto.TextInput("hello")}) + status(t, prepared, "started") + var turn *controlledTurn + select { + case turn = <-executor.turn: + case <-time.After(3 * time.Second): + t.Fatal("no native turn") + } + close(turn.allowCancel) + // Work has started. Losing observation cannot establish its result even + // when Runtime subsequently settles cleanup. + f.conn.Close() + select { + case <-f.core.Closed(): + case <-time.After(3 * time.Second): + t.Fatal("gateway stayed connected") + } + for env := range run.Events { + if env.Type == proto.TypeError || env.Type == proto.TypeDone { + t.Fatal("disconnect manufactured execution failure/completion") + } + } + if !errors.Is(run.Err(), gateway.ErrSessionClosed) { + t.Fatalf("disconnect outcome: %v", run.Err()) + } + conn, err := dial(t, f.endpoint, proto.Version) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + core, err := f.reg.WaitForDevice(t.Context(), "runtime", 3*time.Second) + if err != nil { + t.Fatal(err) + } + defer core.Close("test done") + if core == f.core || f.reg.LookupRun("run") != nil { + t.Fatal("reconnect inherited stale Run ownership") + } + select { + case env := <-conn.Recv(): + t.Fatalf("reconnect replayed work for old handle %s: %+v", ready.Handle, env) + case <-time.After(30 * time.Millisecond): + } + select { + case <-f.stopped: + if f.shutdownErr != nil || executor.closes.Load() != 1 { + t.Fatalf("disconnect cleanup: %v, closes=%d", f.shutdownErr, executor.closes.Load()) + } + case <-time.After(3 * time.Second): + t.Fatal("disconnect cleanup did not settle") + } + if executor.starts.Load() != 1 { + t.Fatal("reconnect replayed input") + } +} diff --git a/apps/parsar-daemon/internal/dispatch/interaction_decisions.go b/apps/parsar-daemon/internal/dispatch/interaction_decisions.go index c4ead6f33..711593393 100644 --- a/apps/parsar-daemon/internal/dispatch/interaction_decisions.go +++ b/apps/parsar-daemon/internal/dispatch/interaction_decisions.go @@ -266,9 +266,6 @@ func (r *Router) indexPermissionFrame(s *sessionState, env proto.Envelope) { return } requestID := strings.TrimSpace(p.RequestID) - if requestID == "" { - requestID = strings.TrimSpace(env.ID) - } if requestID == "" { return } diff --git a/apps/parsar-daemon/internal/dispatch/router_test.go b/apps/parsar-daemon/internal/dispatch/router_test.go index 028d6dbc2..a636c622e 100644 --- a/apps/parsar-daemon/internal/dispatch/router_test.go +++ b/apps/parsar-daemon/internal/dispatch/router_test.go @@ -436,8 +436,7 @@ func TestPromptForUserChoiceDecisionRoutesToSession(t *testing.T) { // payload.AskID to seed askIndex. askEnv := mustEnv(t, proto.TypePromptForUserChoice, "run_ask", proto.PromptForUserChoicePayload{ AskID: "ask_abcd1234", - Question: "?", - Options: []proto.PromptForUserChoiceOption{{Label: "yes"}, {Label: "no"}}, + Questions: []proto.PromptForUserChoiceQuestion{{Question: "?", Options: []proto.PromptForUserChoiceOption{{Label: "yes"}, {Label: "no"}}}}, ToolUseID: "toolu_42", }) sess.out <- askEnv @@ -492,8 +491,7 @@ func TestPromptForUserChoiceDecisionClearsIndexOnAgentUnknown(t *testing.T) { askEnv := mustEnv(t, proto.TypePromptForUserChoice, "run_ask_u", proto.PromptForUserChoicePayload{ AskID: "ask_xxxxxxxx", - Question: "?", - Options: []proto.PromptForUserChoiceOption{{Label: "yes"}}, + Questions: []proto.PromptForUserChoiceQuestion{{Question: "?", Options: []proto.PromptForUserChoiceOption{{Label: "yes"}}}}, ToolUseID: "toolu_y", }) sess.out <- askEnv diff --git a/apps/parsar-daemon/internal/transport/ws.go b/apps/parsar-daemon/internal/transport/ws.go index 8ea9d2f5f..93f7f95f4 100644 --- a/apps/parsar-daemon/internal/transport/ws.go +++ b/apps/parsar-daemon/internal/transport/ws.go @@ -33,7 +33,7 @@ type DialOptions struct { Credential string // DaemonVersion is the X.Y.Z string used for - // proto.VersionCompatible. Mismatches close at upgrade time (4126). + // proto.VersionCompatible. Mismatches reject the HTTP upgrade with 426. DaemonVersion string // HandshakeTimeout caps WS upgrade wait. Zero → 10s. @@ -192,7 +192,7 @@ func (c *Conn) Close() error { // StartHeartbeats kicks off a ticker that calls payloadFn every // interval and Sends the resulting HeartbeatPayload. Returns // immediately. Caller-controlled because the heartbeat carries fields -// (active_requests, claude_available) only the agent layer knows. +// (active_requests, supported_agent_kinds) only the agent layer knows. // Nil logger falls back to log.Bg(). func (c *Conn) StartHeartbeats(parentCtx context.Context, interval time.Duration, payloadFn func() proto.HeartbeatPayload, logger *slog.Logger) { if interval <= 0 || payloadFn == nil { diff --git a/contracts/agents-api/message-input.md b/contracts/agents-api/message-input.md index 7d7f70f1f..89142914f 100644 --- a/contracts/agents-api/message-input.md +++ b/contracts/agents-api/message-input.md @@ -74,7 +74,7 @@ before. ## Runtime boundary -Private wire 0.7.0 uses `MessageInput` for initial requests, prepared start and +Private wire 0.8.0 uses `MessageInput` for initial requests, prepared start and active steering. Each message contains ordered `InputContent` parts, also reused by function-result content. Core does not download, transcode or repair media. The separate Claude bridge reports protocol 2; daemon readiness rejects protocol 1. diff --git a/docs/runtime-protocol.md b/docs/runtime-protocol.md new file mode 100644 index 000000000..df3cc515d --- /dev/null +++ b/docs/runtime-protocol.md @@ -0,0 +1,226 @@ +# Core–Runtime protocol + +This is the integration entry point for a Runtime that executes work for Core. +The wire definitions live once in +[`internal/agentdaemon/proto`](../internal/agentdaemon/proto); +Core's [gateway](../internal/agentdaemon/gateway) and the reference Runtime's +[dispatcher](../apps/parsar-daemon/internal/dispatch) both use them. +The [machine HTTP API](../contracts/agents-api/runtime.openapi.yaml) describes +registration and connection endpoints. This document defines the meaning and +ordering of the messages after connection; it does not replace the typed payloads. + +Use this protocol for hosted and self-hosted Runtime implementations. A new +Harness implements the [adapter contract](../contracts/agents-api/harness-onboarding.md) +behind the Runtime registry. Do not add a Core orchestration branch named after +the Harness, operating system or Sandbox Provider. + +## Ownership and connection + +Core owns durable Session, Turn, input and Environment records, scheduling and +reconciliation. Runtime owns native Executors, active Turns, transfer state and +cleanup until settlement. A Sandbox Provider owns placement and the surrounding +compute lifecycle. Releasing an execution admission or closing an Executor does +not delete, suspend or reclaim a sandbox. + +An installed daemon runs with the authority of the user who starts it. The +protocol does not make the daemon a tool, filesystem or network isolation +boundary. Core-managed Docker, E2B or other sandboxes provide the outer isolation. +A logical workspace, native tool configuration, advertised capability or successful +preparation is not proof of containment. + +1. Obtain the appropriate machine credential using the documented registration + flow. Applications use Project API keys; operators use Core keys. Neither is + a Runtime connection credential. +2. Bootstrap through `/api/v1/agent-daemon/bootstrap` with an Authorization Bearer + header and device identity. Use its authenticated connection URL. +3. Dial the reverse WebSocket at `/api/v1/agent-daemon/ws`, supplying `device_id` + and `version` query parameters and the Bearer header. Never put credentials + in a URL, payload log or trace. +4. Send an immediate heartbeat, then continue at the configured interval. + Declare `supported_agent_kinds`, availability and capabilities explicitly. + An absent first heartbeat means capabilities are unknown; an omitted kind in + a received heartbeat means it is not advertised. Neither permits inference. +5. Dispatch ordered JSON envelopes over the connection. Heartbeats establish + liveness only, not execution progress or a receipt for earlier messages. + +The wire version is [`proto.Version`](../internal/agentdaemon/proto/version.go), +independent of the Runtime build version reported in heartbeats. Core accepts +only an exact match, including the patch component. A mismatch returns HTTP 426 +`incompatible_version` before dispatch; the daemon treats it as permanent and +stops reconnecting. Deploy matching peers together. Removed fields, inferred +Claude availability and old interaction shapes have no compatibility path. + +Each physical connection has fresh routing, admission handles and transfer state. +A newer connection replaces the previous device connection. Core fences owner +leases and evicts old Run/interaction routes; the new connection does not inherit +them. A valid credential and connection are not authority to choose another +Session or Environment binding. + +## Envelope and identity + +Every data frame is one JSON +[`Envelope`](../internal/agentdaemon/proto/envelope.go): +`type`, type-dependent `id`, typed `payload`, and optional W3C `trace`. +Trace is diagnostic correlation only; missing or invalid trace data creates a +local trace and never changes ownership. Do not use a trace ID as a request ID. + +| Identity | Scope and meaning | +| --- | --- | +| Device ID and connection | Authenticated Runtime routing and connection ownership | +| Session ID / Environment ID | Core-owned configuration and workspace binding; canonical UUIDs where required by the payload validator | +| Executor ID | Runtime-owned native resource, potentially retained across settled Turns with identical configuration | +| Preparation request ID | `Envelope.id` for prepare/start/release/status; distinct from a Run | +| Admission handle | Runtime-generated reservation, valid only on its accepting connection | +| Run ID | Execution attempt; `Envelope.id` for output, cancellation, active input and functions | +| Interaction ID | `permission_request.payload.request_id` or `prompt_for_user_choice.payload.ask_id`; these request envelopes still carry the Run ID | +| Delivery ID / input ID / call ID | Resolve attempt, active-input receipt and native function identity respectively; never interchangeable | +| Transfer ID / suspension ID | Connection-local transfer correlation / persisted suspension-attempt fencing | + +Decision and permission-cancel envelopes use the interaction ID. Cancellation and +function-result acknowledgements use the Run ID. All application decision +receipts additionally match the delivery ID. A reply without the required +correlation cannot establish acceptance. + +## Message families + +Use the linked source definitions for required fields, validators, limits and +finite error categories. There is no parallel payload schema to keep in sync. + +| Core → Runtime | Runtime → Core | Definition | +| --- | --- | --- | +| `runtime_prepare` | `runtime_prepare_result` | [Initialization and capability transfer](../internal/agentdaemon/proto/runtime_prepare.go) | +| `execution_prepare`, `execution_start`, `execution_release` | `preparation_status` | [Execution admission](../internal/agentdaemon/proto/preparation.go) | +| `prompt_request`, `prompt_cancel`, `device_shutdown` | `delta`, `thinking`, `output_message`, `tool_call`, `usage`, `error`, `done`, `heartbeat` | [Requests](../internal/agentdaemon/proto/outbound.go), [events and capabilities](../internal/agentdaemon/proto/inbound.go) | +| `permission_decision`, `prompt_for_user_choice_decision` | `permission_request`, `permission_cancel`, `prompt_for_user_choice`, `interaction_decision_ack` | [Requests](../internal/agentdaemon/proto/outbound.go), [interactions](../internal/agentdaemon/proto/inbound.go) | +| `prompt_steer` | `prompt_steer_ack` | [Active input receipts](../internal/agentdaemon/proto/steering.go) | +| `function_result` | `function_call`, `interaction_decision_ack` | [Function calls](../internal/agentdaemon/proto/functions.go) | +| `workspace_read`, `workspace_write`, `workspace_export` | Matching `*_result` | [Read](../internal/agentdaemon/proto/workspace_read.go), [write](../internal/agentdaemon/proto/workspace_write.go), [export](../internal/agentdaemon/proto/workspace_export.go) | +| `environment_quiesce`, `environment_resume` | `environment_quiesced`, `environment_resumed` | [Suspension fencing](../internal/agentdaemon/proto/suspend.go) | + +Initial and active input share [ordered MessageInput](../internal/agentdaemon/proto/message_input.go). +Adapters preserve message/content order and explicitly reject unsupported content. +Usage frames and the final usage snapshot replace earlier cumulative snapshots; +do not add them. An absent measurement is unknown, not zero. + +## Preparation and execution order + +Hosted and self-hosted execution use the same preparation semantics. Placement +selects a connection and Runtime-owned workspace; Harness adapters perform native +configuration. The existing `prompt_request` message remains a direct execution +operation, not a fallback after failed prepared execution. + +Environment initialization uses `runtime_prepare`. For files or archives, send +`begin`, wait for `ready`, send ordered chunks and await matching `received` +offsets, then `commit` and await `completed`. Initialization and finalization +have typed headers without file data. Validate the expected outcome, offset, +size and finite error code with the shared validator. Only one transfer is +allowed per connection. A chunk receipt confirms staged bytes, not installation. +A completed commit confirms that operation, not that a future Turn has executed. + +For an execution Turn: + +1. Subscribe to preparation status before sending `execution_prepare` with + immutable Session configuration and no Run input. +2. `preparing` means the Runtime owns preparation; `ready` supplies the Executor + ID, admission handle, revision and expiry. Neither submits user input. +3. Subscribe to the Run before sending `execution_start` with that Executor, + handle, Run ID and ordered input. A valid start transfers the reservation + once. `started` confirms transfer to the Turn, not Turn completion; output + can race status delivery and must already have a subscriber. +4. Consume Run events until a native terminal outcome or loss of observation. + An execution error is followed by `done` to close that execution stream. + `done` closes the stream; preceding errors remain part of its outcome. +5. On abandonment before start, send `execution_release`. After ownership + transfers to a Run, use `prompt_cancel`; releasing the old handle cannot + cancel its successor. + +Preparation observations use monotonically increasing revisions per handle. +Ignore older or repeated revisions; do not apply a status for a different handle. +Typical transitions are `preparing → ready → starting → started`, or termination +by `released`, `expired`, or `failed`. A `rejected` control operation has an +`operation` and error code but does not replace the resource's current revision. +Read-only workspace preparation cannot start a Turn. Expiry does not remove the +Runtime's obligation to settle cleanup. + +## What each acknowledgement proves + +| Observation | Proven fact | +| --- | --- | +| Core gateway `Send` returns nil | Envelope entered the local send queue | +| Runtime transport `Send` returns nil | WebSocket write completed locally | +| Transfer `received`, active-input `written` | Defined receive/write phase occurred; native execution or consumption is unconfirmed | +| Preparation `preparing` / `ready` | Preparation accepted / reservation ready, with no submitted Run input | +| Preparation `started` | Admission transferred to the identified Turn | +| Active-input `accepted` | Adapter confirmed native consumption under its declared receipt semantics | +| `interaction_decision_ack.applied=true` | Identified operation settled; cancellation additionally requires native settlement | +| `done` and preceding execution events | Execution stream completed with its observed outcome | + +There is no generic receipt for every envelope. A successful send is not proof +that the peer received, accepted or completed a request. Process exit, a stop +signal and a canceled local context do not prove successful cancellation. +A `done` frame may also close a settled cancellation stream; it does not +override the cancellation receipt or imply successful execution. Cancellation +receipts may retain partial content, native identity and usage in `outcome` +even when no `done` is published. Failure to obtain settlement must +remain failed or unknown; it cannot become `applied=true`. + +## Failures, retries and cleanup + +Transport and execution outcomes are separate. Core's only Run subscription +entry point is `SubscribeDurable`; inspect `Subscription.Err()` when its event +channel closes. Disconnection and subscriber overflow close it with an explicit +observation error, without fabricating `error` or `done`. Core retains durable +truth and reconciles from confirmed facts. Runtime retains cleanup ownership +until native work, input receipts, interactions and child work have settled. + +The public Turn status is a separate, existing projection: +[`execution/delivery.go`](../services/agents-api/internal/execution/delivery.go) +records an unsuccessful orchestration attempt as `failed`, including +`delivery_unknown` after an unconfirmed send and `event_stream_incomplete` +after subscription failure. A closed subscription can replace the send reason +with `event_stream_incomplete`; both retain an unknown native effect. +This public `failed` status is not proof that the Harness failed, that no side +effect occurred or that cleanup completed. Native error classification comes +only from observed Runtime error frames. Consumers must keep the observation +reason and any native evidence distinct; this contract does not add a public +`unknown` status or change the existing Turn state machine. + +| Condition | Required responsibility | +| --- | --- | +| Unsupported capability or invalid binding | Reject explicitly before starting the unsupported operation; do not select another Harness | +| Confirmed preparation/execution failure | Preserve the finite error category and any observed result; Runtime settles its resources | +| Deadline or connection loss after dispatch | Caller has an unknown effect unless an application receipt proves otherwise; do not convert it to execution failure | +| Reconnection | Reestablish transport and capability advertisement; do not replay input, initialization, transfers or unresolved mutations | +| Duplicate preparation/start | Existing connection-local identity and fingerprint rules apply; a conflicting request rejects and an old handle cannot start replacement work | +| Duplicate input/function/decision | Use that family's existing receipt identity and conflict rules; no transport-wide deduplication or exactly-once promise exists | +| Cleanup failure | Retain resource ownership and report unconfirmed cleanup; a resource is not reusable merely because a waiter timed out | +| Lost cancellation receipt | Cancellation may have happened; lack of receipt cannot establish success or authorize another execution | + +Only retry operations whose own contract establishes that retry is safe. Retired +preparation request IDs may eventually allocate a new handle, so they are not +durable idempotency keys. Native-session resume is an explicit operation with +verified identity, not a response to a socket failure. A transient connection +error permits reconnecting the channel; authentication/version rejection requires +operator correction. Cleanup of an Executor remains separate from the Sandbox +Provider's confirmed reclamation of compute. + +## Contract verification + +Run `make check-runtime-contract` from the repository root. It exercises the +shared wire validators, gateway, transport and dispatcher, plus +[real WebSocket contract scenarios](../apps/parsar-daemon/internal/contracttest/wire_test.go) +using a controlled Harness adapter, plus the [observation-result regression](../services/agents-api/internal/execution/runtime_protocol_test.go). It requires no model credentials or external +sandbox. These tests are also included in `make check` through `check-go` and +`check-agents-api`. + +The suite checks incompatible versions, preparation failure, cancellation +settlement, connection loss without invented terminal events, reconnect without +replay, stale/duplicate handles and receipts, cleanup failures, bounded transfer +validation and resource ownership after timeout. Existing detailed fault +injection remains next to the owning gateway/dispatcher implementation. + +For another Runtime or Harness, reuse these protocol sequences and assertions, +then run its own native acceptance for the capabilities it advertises. A passing +controlled-adapter test establishes the transport contract, not native Harness +behavior, OS support, provider authentication or sandbox isolation. Update the +shared types, this guide and the contract checks together when semantics change. diff --git a/internal/agentdaemon/device/state.go b/internal/agentdaemon/device/state.go index 4d9ac1a6a..42b9b69ea 100644 --- a/internal/agentdaemon/device/state.go +++ b/internal/agentdaemon/device/state.go @@ -72,7 +72,6 @@ type KindCapabilities struct { ToolObservations bool `json:"tool_observations,omitempty"` EnvironmentNone bool `json:"environment_none,omitempty"` LocalEnvironment bool `json:"local_environment,omitempty"` - LocalEnvironmentNetworkPolicy bool `json:"local_environment_network_policy,omitempty"` Preparation bool `json:"preparation,omitempty"` WorkspaceReadPreparation bool `json:"workspace_read_preparation,omitempty"` WorkspaceOutputExport bool `json:"workspace_output_export,omitempty"` diff --git a/internal/agentdaemon/gateway/handler.go b/internal/agentdaemon/gateway/handler.go index cbfb56f1e..9783d5faf 100644 --- a/internal/agentdaemon/gateway/handler.go +++ b/internal/agentdaemon/gateway/handler.go @@ -101,7 +101,7 @@ func NewHandler(cfg HandlerConfig) *Handler { // WS is the websocket upgrade entry point. Errors before the upgrade // return JSON 4xx; errors during the WS read loop fall to Session.Close -// which fans synthetic error/done to every active subscriber. +// which closes subscriptions with a transport error, without execution events. // // @Summary Agent-daemon WebSocket upgrade // @Description Long-lived duplex channel for daemon runtimes. Authenticated by the runner bearer in the HTTP Authorization header before upgrade. diff --git a/internal/agentdaemon/gateway/preparation.go b/internal/agentdaemon/gateway/preparation.go index 8f09abd56..286624fc2 100644 --- a/internal/agentdaemon/gateway/preparation.go +++ b/internal/agentdaemon/gateway/preparation.go @@ -25,7 +25,7 @@ func (s *Session) SubscribePreparation(requestID string) (*Subscription, error) return nil, errors.New("agentdaemon gateway: invalid, duplicate or excess preparation subscription") } ch := make(chan proto.Envelope, 16) - sub := &Subscription{Events: ch, ch: ch, durable: true} + sub := &Subscription{Events: ch, ch: ch} s.preparations[requestID] = &preparationSubscription{sub: sub} return sub, nil } diff --git a/internal/agentdaemon/gateway/preparation_test.go b/internal/agentdaemon/gateway/preparation_test.go index ed6ea3acf..496a0994b 100644 --- a/internal/agentdaemon/gateway/preparation_test.go +++ b/internal/agentdaemon/gateway/preparation_test.go @@ -77,8 +77,8 @@ func TestPreparationCloseAndOverflowDoNotInventRunEvents(t *testing.T) { } func TestPreparationCapabilitySurvivesHeartbeatMapping(t *testing.T) { - kinds := deviceKindsFromHeartbeat(proto.HeartbeatPayload{SupportedAgentKinds: []proto.SupportedAgentKind{{Kind: "codex", Available: true, Capabilities: proto.AgentKindCapabilities{Preparation: true, LocalEnvironment: true, LocalEnvironmentNetworkPolicy: true, WorkspaceReadPreparation: true, NativeSessionRecovery: true}}}}) - if len(kinds) != 1 || (!kinds[0].Capabilities.Preparation || !kinds[0].Capabilities.LocalEnvironment || !kinds[0].Capabilities.LocalEnvironmentNetworkPolicy || !kinds[0].Capabilities.WorkspaceReadPreparation || !kinds[0].Capabilities.NativeSessionRecovery) { + kinds := deviceKindsFromHeartbeat(proto.HeartbeatPayload{SupportedAgentKinds: []proto.SupportedAgentKind{{Kind: "codex", Available: true, Capabilities: proto.AgentKindCapabilities{Preparation: true, LocalEnvironment: true, WorkspaceReadPreparation: true, NativeSessionRecovery: true}}}}) + if len(kinds) != 1 || (!kinds[0].Capabilities.Preparation || !kinds[0].Capabilities.LocalEnvironment || !kinds[0].Capabilities.WorkspaceReadPreparation || !kinds[0].Capabilities.NativeSessionRecovery) { t.Fatal("preparation capability lost") } } diff --git a/internal/agentdaemon/gateway/session.go b/internal/agentdaemon/gateway/session.go index edd779e1c..f74b7e5fd 100644 --- a/internal/agentdaemon/gateway/session.go +++ b/internal/agentdaemon/gateway/session.go @@ -205,8 +205,6 @@ func (s *Session) LastSeen() time.Time { // AgentKindStatus returns the latest advertised descriptor for kind. // found=false means the daemon has not advertised that kind; snapshotKnown // distinguishes "no heartbeat yet" from "heartbeat arrived and omitted it". -// Before the first heartbeat, legacy Claude Code behavior is preserved so -// older daemons can still receive claude_code prompt_requests immediately. func (s *Session) AgentKindStatus(kind string) (info device.SupportedAgentKind, found bool, snapshotKnown bool) { kind = strings.TrimSpace(kind) if kind == "" { @@ -218,9 +216,6 @@ func (s *Session) AgentKindStatus(kind string) (info device.SupportedAgentKind, copy(kinds, s.supportedKinds) s.kindsMu.RUnlock() if !seen { - if kind == "claude_code" { - return legacyClaudeCodeKind(), true, false - } return device.SupportedAgentKind{}, false, false } for _, candidate := range kinds { @@ -240,34 +235,20 @@ func (s *Session) setSupportedAgentKinds(kinds []device.SupportedAgentKind) { s.kindsMu.Unlock() } -func legacyClaudeCodeKind() device.SupportedAgentKind { - return device.SupportedAgentKind{ - Kind: "claude_code", - Available: true, - Capabilities: device.KindCapabilities{ - Streaming: true, - Permissions: true, - Usage: true, - Resume: true, - }, - } -} - -// Close tears the session down: closes the WS, drains subscribers -// with a synthetic error+done pair, and deregisters. Idempotent. +// Close closes the transport and subscriptions with ErrSessionClosed, then +// releases connection ownership. It establishes no execution outcome. Idempotent. func (s *Session) Close(reason string) { s.closeOnce.Do(func() { s.stopReceiptTimer() close(s.closed) _ = s.conn.Close() - // Synthetic error + done so the connector's translation loop - // sees a clean EOF and unsubscribes naturally. + // Transport failure must remain distinct from native execution facts. s.subsMu.Lock() subs := s.subs s.subs = map[string]*Subscription{} s.subsMu.Unlock() for runID, sub := range subs { - s.closeSubscription(runID, sub, reason) + s.closeSubscription(sub) s.reg.DetachRun(runID) } s.reg.Deregister(s) @@ -291,24 +272,6 @@ func (s *Session) CloseWithCode(code int, reason string) { s.Close(reason) } -func (s *Session) deliverSynthetic(runID string, ch chan proto.Envelope, reason string) { - if reason == "" { - reason = "device disconnected" - } - errEnv, _ := proto.NewEnvelope(proto.TypeError, runID, proto.ErrorPayload{Error: reason}) - doneEnv, _ := proto.NewEnvelope(proto.TypeDone, runID, proto.DonePayload{}) - // Non-blocking — drop rather than hang the close path on a - // wedged subscriber. - select { - case ch <- errEnv: - default: - } - select { - case ch <- doneEnv: - default: - } -} - // Send queues an envelope for the WS write loop. Returns ErrSessionClosed // if the session has shut down, or context.DeadlineExceeded if the send // queue is full for longer than ctx's timeout. @@ -540,21 +503,6 @@ func (s *Session) handleHeartbeat(env proto.Envelope) { } func deviceKindsFromHeartbeat(p proto.HeartbeatPayload) []device.SupportedAgentKind { - if len(p.SupportedAgentKinds) == 0 { - if !p.ClaudeAvailable { - return nil - } - return []device.SupportedAgentKind{{ - Kind: "claude_code", - Available: true, - Capabilities: device.KindCapabilities{ - Streaming: true, - Permissions: true, - Usage: true, - Resume: true, - }, - }} - } out := make([]device.SupportedAgentKind, 0, len(p.SupportedAgentKinds)) for _, info := range p.SupportedAgentKinds { out = append(out, device.SupportedAgentKind{ @@ -575,7 +523,6 @@ func deviceKindsFromHeartbeat(p proto.HeartbeatPayload) []device.SupportedAgentK ToolObservations: info.Capabilities.ToolObservations, EnvironmentNone: info.Capabilities.EnvironmentNone, LocalEnvironment: info.Capabilities.LocalEnvironment, - LocalEnvironmentNetworkPolicy: info.Capabilities.LocalEnvironmentNetworkPolicy, Preparation: info.Capabilities.Preparation, WorkspaceReadPreparation: info.Capabilities.WorkspaceReadPreparation, WorkspaceOutputExport: info.Capabilities.WorkspaceOutputExport, @@ -631,11 +578,9 @@ func (s *Session) dispatch(env proto.Envelope) { requestID = strings.TrimSpace(p.RequestID) } if requestID == "" { - requestID = strings.TrimSpace(env.ID) - } - if requestID != "" { - s.reg.AttachPermission(requestID, s) + return } + s.reg.AttachPermission(requestID, s) case proto.TypePermissionCancel: if env.ID != "" { s.reg.DetachPermission(env.ID) @@ -671,9 +616,7 @@ func (s *Session) dispatch(env proto.Envelope) { // All run-correlated frames fan to the matching subscriber. Current // permission and prompt-for-user-choice frames keep their interaction ID - // in the payload so Envelope.ID remains the run ID. Legacy permission - // frames put the permission ID in Envelope.ID; those are still indexed - // above but cannot be correlated to a run subscriber. + // in the payload so Envelope.ID remains the run ID. if env.ID == "" { return } diff --git a/internal/agentdaemon/gateway/session_test.go b/internal/agentdaemon/gateway/session_test.go index 30fc4396f..74f7b4ad6 100644 --- a/internal/agentdaemon/gateway/session_test.go +++ b/internal/agentdaemon/gateway/session_test.go @@ -206,7 +206,8 @@ func TestSession_DispatchDeliversToSubscriber(t *testing.T) { sess.Start() defer sess.Close("test done") - ch, err := sess.Subscribe("run-1") + sub, err := sess.SubscribeDurable("run-1") + ch := sub.Events if err != nil { t.Fatalf("Subscribe: %v", err) } @@ -231,7 +232,8 @@ func TestSession_DoneFrameAutoUnsubscribes(t *testing.T) { sess.Start() defer sess.Close("test done") - ch, _ := sess.Subscribe("run-1") + sub, _ := sess.SubscribeDurable("run-1") + ch := sub.Events env, _ := proto.NewEnvelope(proto.TypeDone, "run-1", proto.DonePayload{Content: "ok"}) raw, _ := jsonMarshal(env) conn.Feed(raw) @@ -265,7 +267,8 @@ func TestSession_PermissionRequestIndexedInRegistry(t *testing.T) { sess.Start() defer sess.Close("test done") - ch, err := sess.Subscribe("run-1") + sub, err := sess.SubscribeDurable("run-1") + ch := sub.Events if err != nil { t.Fatalf("subscribe: %v", err) } @@ -299,34 +302,27 @@ func TestSession_PermissionRequestIndexedInRegistry(t *testing.T) { } } -func TestSession_CloseFansSyntheticErrorAndDone(t *testing.T) { - reg := NewRegistry() - conn := newFakeConn() - sess := NewSession(conn, "dev-1", "wks-1", "0.1.0", reg, nil) - sess.Start() - - ch, _ := sess.Subscribe("run-1") +func TestSession_CloseReportsUnknownWithoutExecutionEvents(t *testing.T) { + sess := NewSession(newFakeConn(), "device", "tenant", proto.Version, NewRegistry(), nil) + sub, err := sess.SubscribeDurable("run") + if err != nil { + t.Fatal(err) + } sess.Close("simulated drop") + for env := range sub.Events { + t.Fatalf("disconnect fabricated execution event: %s", env.Type) + } + if !errors.Is(sub.Err(), ErrSessionClosed) { + t.Fatalf("missing transport error: %v", sub.Err()) + } +} - gotErr, gotDone := false, false - deadline := time.After(2 * time.Second) - for !gotErr || !gotDone { - select { - case env, ok := <-ch: - if !ok { - if !gotErr || !gotDone { - t.Fatalf("channel closed before delivering synthetic error+done (gotErr=%v gotDone=%v)", gotErr, gotDone) - } - return - } - switch env.Type { - case proto.TypeError: - gotErr = true - case proto.TypeDone: - gotDone = true - } - case <-deadline: - t.Fatalf("synthetic frames not delivered (gotErr=%v gotDone=%v)", gotErr, gotDone) +func TestSession_NoCapabilitiesBeforeHeartbeat(t *testing.T) { + sess := NewSession(newFakeConn(), "device", "tenant", proto.Version, NewRegistry(), nil) + defer sess.Close("test done") + for _, kind := range []string{"claude_code", "codex", "contract"} { + if info, found, known := sess.AgentKindStatus(kind); found || known || info.Available { + t.Fatalf("unadvertised engine was available: %+v", info) } } } @@ -443,33 +439,45 @@ func TestSession_HeartbeatPersistsSupportedAgentKinds(t *testing.T) { } } -func TestSession_HeartbeatInfersClaudeCodeFromLegacyFlag(t *testing.T) { +func TestSession_HeartbeatDoesNotInferCapabilities(t *testing.T) { reg := NewRegistry() conn := newFakeConn() heartbeat := newFakeHeartbeatStore() - sess := NewSession(conn, "dev-legacy", "wks-1", "0.1.0", reg, nil) + sess := NewSession(conn, "device", "tenant", proto.Version, reg, nil) sess.heartbeat = heartbeat sess.Start() defer sess.Close("test done") - env, _ := proto.NewEnvelope(proto.TypeHeartbeat, "", proto.HeartbeatPayload{ - Timestamp: 1710000100, - DaemonVersion: "0.1.0-old", - ClaudeAvailable: true, - }) - raw, _ := jsonMarshal(env) - conn.Feed(raw) - + conn.Feed([]byte(`{"type":"heartbeat","payload":{"ts":1710000100,"claude_available":true}}`)) got := heartbeat.waitDaemonHeartbeat(t) - if len(got.SupportedAgentKinds) != 1 { - t.Fatalf("SupportedAgentKinds len = %d, want 1: %#v", len(got.SupportedAgentKinds), got.SupportedAgentKinds) + if len(got.SupportedAgentKinds) != 0 { + t.Fatalf("undeclared capabilities inferred: %#v", got.SupportedAgentKinds) + } +} + +func TestSession_PermissionRequiresPayloadIdentity(t *testing.T) { + reg := NewRegistry() + sess := NewSession(newFakeConn(), "device", "tenant", proto.Version, reg, nil) + defer sess.Close("test done") + sub, err := sess.SubscribeDurable("run") + if err != nil { + t.Fatal(err) + } + env, _ := proto.NewEnvelope(proto.TypePermissionRequest, "run", proto.PermissionRequestPayload{Tool: "test"}) + sess.dispatch(env) + if _, err := reg.LookupPermission("run"); !errors.Is(err, ErrPermissionNotRegistered) { + t.Fatal("run ID was treated as an interaction ID") + } + if len(sub.Events) != 0 { + t.Fatal("invalid permission request was forwarded") } - claude := got.SupportedAgentKinds[0] - if claude.Kind != "claude_code" || !claude.Available || !claude.Capabilities.Streaming || !claude.Capabilities.Permissions || !claude.Capabilities.Usage || !claude.Capabilities.Resume { - t.Fatalf("legacy claude_available fallback not inferred: %#v", claude) + env, _ = proto.NewEnvelope(proto.TypePermissionRequest, "run", proto.PermissionRequestPayload{RequestID: "permission", Tool: "test"}) + sess.dispatch(env) + if got, err := reg.LookupPermission("permission"); err != nil || got != sess { + t.Fatalf("declared interaction identity was not registered: %v", err) } - if claude.Capabilities.Steering { - t.Fatal("legacy daemon must not advertise steering") + if got := <-sub.Events; got.ID != "run" { + t.Fatal("run correlation lost") } } diff --git a/internal/agentdaemon/gateway/subscription.go b/internal/agentdaemon/gateway/subscription.go index 71bc9c1fc..4dd79189e 100644 --- a/internal/agentdaemon/gateway/subscription.go +++ b/internal/agentdaemon/gateway/subscription.go @@ -11,12 +11,11 @@ import ( var ErrSubscriberOverflow = errors.New("execution subscriber buffer overflow") type Subscription struct { - Events <-chan proto.Envelope - ch chan proto.Envelope - mu sync.Mutex - err error - closed bool - durable bool + Events <-chan proto.Envelope + ch chan proto.Envelope + mu sync.Mutex + err error + closed bool } func (s *Subscription) Err() error { @@ -32,30 +31,14 @@ func (s *Subscription) closeLocked(err error) { } } -// Subscribe retains the product's best-effort stream behavior. -func (s *Session) Subscribe(runID string) (<-chan proto.Envelope, error) { - sub, err := s.subscribe(runID, false) - if err != nil { - return nil, err - } - return sub.Events, nil -} - -// SubscribeDurable fails the subscription on overflow instead of losing events silently. +// SubscribeDurable reports transport loss and overflow separately from native +// execution events. Callers must inspect Err after Events closes. func (s *Session) SubscribeDurable(runID string) (*Subscription, error) { - return s.subscribe(runID, true) -} - -func (s *Session) subscribe(runID string, durable bool) (*Subscription, error) { if runID == "" { return nil, fmt.Errorf("agentdaemon gateway: Subscribe requires non-empty runID") } - capacity := 32 - if durable { - capacity = 256 - } - ch := make(chan proto.Envelope, capacity) - sub := &Subscription{Events: ch, ch: ch, durable: durable} + ch := make(chan proto.Envelope, 256) + sub := &Subscription{Events: ch, ch: ch} s.subsMu.Lock() defer s.subsMu.Unlock() if s.IsClosed() { @@ -83,15 +66,12 @@ func (s *Session) Unsubscribe(runID string) { s.reg.DetachRun(runID) } -func (s *Session) closeSubscription(runID string, sub *Subscription, reason string) { +func (s *Session) closeSubscription(sub *Subscription) { sub.mu.Lock() defer sub.mu.Unlock() if sub.closed { return } - if !sub.durable { - s.deliverSynthetic(runID, sub.ch, reason) - } sub.closeLocked(ErrSessionClosed) } @@ -110,14 +90,7 @@ func (s *Session) dispatchToSubscriber(env proto.Envelope) { sub.closeLocked(nil) } default: - if sub.durable { - sub.closeLocked(ErrSubscriberOverflow) - } else { - s.log("agentdaemon gateway: subscriber buffer full for run %s, dropping %s", env.ID, env.Type) - if env.Type == proto.TypeDone { - sub.closeLocked(nil) - } - } + sub.closeLocked(ErrSubscriberOverflow) } if sub.closed { delete(s.subs, env.ID) diff --git a/internal/agentdaemon/gateway/subscription_test.go b/internal/agentdaemon/gateway/subscription_test.go index 47ad15473..3c79343ae 100644 --- a/internal/agentdaemon/gateway/subscription_test.go +++ b/internal/agentdaemon/gateway/subscription_test.go @@ -33,28 +33,6 @@ func TestDurableSubscriptionOverflowIsExplicitAndIsolated(t *testing.T) { } } -func TestProductSubscriptionRetainsBestEffortBuffer(t *testing.T) { - s := NewSession(newFakeConn(), "device", "tenant", proto.Version, NewRegistry(), nil) - defer s.Close("test finished") - events, _ := s.Subscribe("product") - for range 33 { - s.dispatchToSubscriber(proto.Envelope{ID: "product", Type: proto.TypeDelta}) - } - if len(events) != 32 { - t.Fatal("product buffer changed") - } - for range 32 { - <-events - } - s.dispatchToSubscriber(proto.Envelope{ID: "product", Type: proto.TypeDone}) - if event := <-events; event.Type != proto.TypeDone { - t.Fatal(event.Type) - } - if _, ok := <-events; ok { - t.Fatal("terminal stream not closed") - } -} - func TestSubscriptionCloseAndDispatchAreSerialized(t *testing.T) { for range 100 { s := NewSession(newFakeConn(), "device", "tenant", proto.Version, NewRegistry(), nil) diff --git a/internal/agentdaemon/proto/envelope.go b/internal/agentdaemon/proto/envelope.go index 73eca4200..da98ad3df 100644 --- a/internal/agentdaemon/proto/envelope.go +++ b/internal/agentdaemon/proto/envelope.go @@ -1,6 +1,7 @@ // Package proto defines the JSON wire format spoken by oac-daemon over // the reverse WebSocket tunnel to the OpenAgentCore server. Both ends import -// this package; adding an event means editing one file here and both +// this package. See docs/runtime-protocol.md for the integration contract. +// Adding an event means updating the shared definition and both // sides at once. // // Topology: daemon dials OUT (firewall-friendly). Every frame is one @@ -13,8 +14,11 @@ // Envelope.ID correlation: // - prompt_request / prompt_cancel: ID = RunID. // - delta / tool_call / usage / error / done: ID = originating RunID. -// - permission_request: ID = daemon-minted "perm_<8hex>"; the matching -// downstream permission_decision echoes it back. +// - permission_request: ID = RunID; payload.request_id is the interaction ID. +// - permission_decision / permission_cancel: ID = interaction ID. +// - execution_prepare / execution_start / execution_release and +// preparation_status: ID = preparation request ID, never RunID. +// - runtime_prepare / runtime_prepare_result: ID = connection-local transfer ID. // - heartbeats carry no ID. package proto @@ -42,7 +46,7 @@ type Envelope struct { // ("00-{32hex trace_id}-{16hex span_id}-{2hex flags}") for the // logical request. Server-issued envelopes carry the gateway's // ctx carrier; daemon-issued envelopes carry the daemon's. Omitted - // when no trace is in scope (heartbeats, legacy clients). + // when no trace is in scope (for example, heartbeats). // Receivers MUST tolerate missing/unparseable values — both mean // "mint a fresh trace locally", never reject. Trace string `json:"trace,omitempty"` diff --git a/internal/agentdaemon/proto/envelope_test.go b/internal/agentdaemon/proto/envelope_test.go index c7388daca..d485d8f7d 100644 --- a/internal/agentdaemon/proto/envelope_test.go +++ b/internal/agentdaemon/proto/envelope_test.go @@ -88,12 +88,17 @@ func TestVersionCompatible(t *testing.T) { ok bool }{ {Version, true}, // exact match - {"0.7.99", true}, // patch drift OK + {"0.8.99", false}, // patch drift NOT OK {"0.6.99", false}, // minor drift NOT OK {"1.0.0", false}, // major drift NOT OK {"", false}, // missing {"garbage", false}, // unparseable - {"0.3", false}, // truncated + {"0.8", false}, // truncated + {"0.8.", false}, + {"0.8.invalid", false}, + {Version + "-dev", false}, + {Version + "+build", false}, + {" " + Version, false}, } for _, tc := range cases { if got := VersionCompatible(tc.client); got != tc.ok { diff --git a/internal/agentdaemon/proto/inbound.go b/internal/agentdaemon/proto/inbound.go index 1060f6515..ed0bedb55 100644 --- a/internal/agentdaemon/proto/inbound.go +++ b/internal/agentdaemon/proto/inbound.go @@ -31,7 +31,7 @@ const ( TypeToolCall = "tool_call" // TypePermissionRequest carries an agent's request for human - // approval. Envelope.ID = "perm_<8hex>" minted by the daemon. + // approval. Envelope.ID is the RunID; payload.request_id identifies the decision. TypePermissionRequest = "permission_request" // TypePermissionCancel signals the agent withdrew an earlier @@ -43,7 +43,7 @@ const ( // answers from a closed list before the agent can continue. Used // to intercept Claude Code's built-in AskUserQuestion tool so the // daemon doesn't deadlock waiting for a tool_result no one will - // send. Envelope.ID = "ask_<8hex>" minted by the daemon. + // send. Envelope.ID is the RunID; payload.ask_id identifies the decision. TypePromptForUserChoice = "prompt_for_user_choice" // TypeInteractionDecisionAck confirms that the daemon-side agent @@ -108,10 +108,8 @@ type ToolCallPayload struct { // approval. RequestID is the daemon-minted handle used to route a later // decision. It lives in the payload because Envelope.ID is the run ID used // by the server gateway to deliver the request to the active run subscriber. -// Readers still accept legacy frames that omit RequestID and put the request -// handle in Envelope.ID. type PermissionRequestPayload struct { - RequestID string `json:"request_id,omitempty"` + RequestID string `json:"request_id"` Tool string `json:"tool"` Title string `json:"title"` Detail string `json:"detail,omitempty"` @@ -161,41 +159,11 @@ type PromptForUserChoiceQuestion struct { // the originating Claude Code tool_use id; empty when the call came // through the control_request channel (CCRequestID then identifies the // daemon-side waiter instead but it doesn't ride on the wire). -// -// The legacy single-question fields (Question / Header / MultiSelect / -// Options) stay on the wire so older server/db snapshots can still be -// decoded. New code writes Questions; readers must call -// EffectiveQuestions to get a unified view across both shapes. type PromptForUserChoicePayload struct { AskID string `json:"ask_id"` - Questions []PromptForUserChoiceQuestion `json:"questions,omitempty"` + Questions []PromptForUserChoiceQuestion `json:"questions"` ToolUseID string `json:"tool_use_id,omitempty"` AutoResolutionMs *uint64 `json:"auto_resolution_ms,omitempty"` - - // Legacy single-question fields — read-only on the new path. Empty - // when Questions is populated. - Question string `json:"question,omitempty"` - Header string `json:"header,omitempty"` - MultiSelect bool `json:"multi_select,omitempty"` - Options []PromptForUserChoiceOption `json:"options,omitempty"` -} - -// EffectiveQuestions returns the question list a consumer should -// render. Prefers the new Questions slice; falls back to the legacy -// single-question fields so old payloads still work after a restart. -func (p PromptForUserChoicePayload) EffectiveQuestions() []PromptForUserChoiceQuestion { - if len(p.Questions) > 0 { - return p.Questions - } - if p.Question == "" && len(p.Options) == 0 { - return nil - } - return []PromptForUserChoiceQuestion{{ - Header: p.Header, - Question: p.Question, - MultiSelect: p.MultiSelect, - Options: p.Options, - }} } // TokenUsage is a complete cumulative measurement for the current execution. @@ -269,7 +237,6 @@ type AgentKindCapabilities struct { ToolObservations bool `json:"tool_observations,omitempty"` EnvironmentNone bool `json:"environment_none,omitempty"` LocalEnvironment bool `json:"local_environment,omitempty"` - LocalEnvironmentNetworkPolicy bool `json:"local_environment_network_policy,omitempty"` Preparation bool `json:"preparation,omitempty"` WorkspaceReadPreparation bool `json:"workspace_read_preparation,omitempty"` WorkspaceOutputExport bool `json:"workspace_output_export,omitempty"` @@ -302,14 +269,11 @@ type SupportedAgentKind struct { Capabilities AgentKindCapabilities `json:"capabilities,omitempty"` } -// HeartbeatPayload is the daemon's liveness ping. supported_agent_kinds -// is preferred over the legacy claude_available flag; old daemons may -// still send only claude_available, and the server infers claude_code -// support. +// HeartbeatPayload advertises only explicit engine descriptors. Missing +// supported_agent_kinds establishes no engine availability or capabilities. type HeartbeatPayload struct { Timestamp int64 `json:"ts"` ActiveRequests int `json:"active_requests"` DaemonVersion string `json:"daemon_version,omitempty"` - ClaudeAvailable bool `json:"claude_available,omitempty"` SupportedAgentKinds []SupportedAgentKind `json:"supported_agent_kinds,omitempty"` } diff --git a/internal/agentdaemon/proto/version.go b/internal/agentdaemon/proto/version.go index 8644ad91e..9725840e6 100644 --- a/internal/agentdaemon/proto/version.go +++ b/internal/agentdaemon/proto/version.go @@ -1,40 +1,11 @@ package proto -// Protocol version. Bump on any existing-payload-shape change (additive -// optional fields don't need a bump). Daemon sends this in the WS -// upgrade query (`version=`) and in the bootstrap HTTP -// response; mismatches fail closed at WS upgrade. -const ( - Version = "0.7.0" -) +// Version identifies the complete Core–Runtime wire contract. Change it when +// removing or changing a payload or its semantics; deploy both endpoints together. +const Version = "0.8.0" -// VersionCompatible returns true when clientVersion's "X.Y" prefix -// exactly matches Version's. Inputs without a patch segment (e.g. -// "0.1", "1") are rejected so a half-set version string never -// spuriously matches. +// VersionCompatible accepts only this contract. Patch drift, prerelease suffixes +// and malformed versions do not select an implicit compatibility path. func VersionCompatible(clientVersion string) bool { - if clientVersion == "" { - return false - } - clientMM, ok := majorMinor(clientVersion) - if !ok { - return false - } - serverMM, _ := majorMinor(Version) - return clientMM == serverMM -} - -// majorMinor returns the "X.Y" prefix of a semver-shaped version. -// Second return is false when the input has no patch segment. -func majorMinor(v string) (string, bool) { - dots := 0 - for i := 0; i < len(v); i++ { - if v[i] == '.' { - dots++ - if dots == 2 { - return v[:i], true - } - } - } - return "", false + return clientVersion == Version } diff --git a/scripts/name-allowlist.json b/scripts/name-allowlist.json index 37c3f3d0d..33861441c 100644 --- a/scripts/name-allowlist.json +++ b/scripts/name-allowlist.json @@ -853,5 +853,10 @@ "path": "apps/parsar-daemon/internal/localworkspace/capability_preparation_test.go", "regex": "\"agents-api-\"", "reason": "AgentStateKey retains the existing daemon/native-session resume identity; capability snapshots bind to the Session UUID carried in that identity." + }, + { + "path": "apps/parsar-daemon/internal/contracttest/wire_test.go", + "regex": "agents-api-session", + "reason": "The controlled Runtime fixture uses the existing Core Session state-key prefix required by execution admission." } ] diff --git a/services/agents-api/internal/execution/runtime_protocol_test.go b/services/agents-api/internal/execution/runtime_protocol_test.go new file mode 100644 index 000000000..02b65a99c --- /dev/null +++ b/services/agents-api/internal/execution/runtime_protocol_test.go @@ -0,0 +1,54 @@ +package execution + +import ( + "context" + "encoding/json" + "testing" + + "github.com/MiniMax-AI-Dev/parsar/internal/agentdaemon/proto" +) + +// Public Turn failure can describe lost orchestration, independently of native +// failure. Late observations must not replace its unknown-effect reason or +// manufacture a native terminal result. +func TestRuntimeProtocolUnknownEffectRetainsObservationFailure(t *testing.T) { + for _, reason := range []string{"delivery_unknown", "event_stream_incomplete"} { + t.Run(reason, func(t *testing.T) { + result := Result{ErrorCode: reason, AppliedThrough: 1} + writer := &recoveringWriter{} + j := journal{store: writer, next: 1} + events := make(chan proto.Envelope, 1) + usage, err := proto.NewEnvelope(proto.TypeUsage, "run", proto.Usage{InputTokens: 7}) + if err != nil { + t.Fatal(err) + } + events <- usage + close(events) + if err := j.drain(events, &result); err != nil { + t.Fatal(err) + } + // A cancellation waiter deadline is not an application receipt. + if err := recordCancellation(context.Background(), &j, cancellationResult{err: context.DeadlineExceeded}, &result); err != nil { + t.Fatal(err) + } + if err := j.flush(context.Background()); err != nil { + t.Fatal(err) + } + raw, err := json.Marshal(result) + if err != nil { + t.Fatal(err) + } + var persisted Result + if err := json.Unmarshal(raw, &persisted); err != nil { + t.Fatal(err) + } + if persisted.ErrorCode != reason || persisted.EngineErrorCode != "" || persisted.EngineHTTPStatus != nil || + persisted.Done.SourceCompletedAtMS != nil || persisted.Done.Usage.InputTokens != 7 || persisted.AppliedThrough != 1 { + t.Fatalf("unknown result changed into native terminal evidence: %+v", persisted) + } + if len(writer.events) != 1 || writer.events[0].Kind != proto.TypeUsage { + t.Fatalf("unknown delivery fabricated completion or cancellation: %+v", writer.events) + } + }) + } +} diff --git a/services/agents-api/internal/execution/support.go b/services/agents-api/internal/execution/support.go index 9982f8f81..37f68a25f 100644 --- a/services/agents-api/internal/execution/support.go +++ b/services/agents-api/internal/execution/support.go @@ -130,9 +130,6 @@ func (p Policy) engineCapabilities(peer *gateway.Session, engine string, snapsho if !caps.Preparation || !caps.LocalEnvironment || !caps.WorkspaceReadPreparation || !caps.WorkspaceOutputExport { return fail("device must advertise local preparation, workspace reads and output export") } - if !caps.LocalEnvironmentNetworkPolicy { - return fail("device must advertise local_environment_network_policy") - } } if snapshot.Environment != nil && snapshot.Environment.Type == "none" && !caps.EnvironmentNone { return fail("device must advertise environment_none") diff --git a/services/agents-api/internal/store/environment_worker_helpers_test.go b/services/agents-api/internal/store/environment_worker_helpers_test.go index 168df067a..a5163e502 100644 --- a/services/agents-api/internal/store/environment_worker_helpers_test.go +++ b/services/agents-api/internal/store/environment_worker_helpers_test.go @@ -26,7 +26,7 @@ func enableWorkerEnvironment(t *testing.T, h *dispatchHarness) { } func workerEnvironmentCapabilities() proto.AgentKindCapabilities { - return proto.AgentKindCapabilities{Streaming: true, Steering: true, DurableTurns: true, DurableInputReceipts: true, EnvironmentNone: true, WebSearchControl: true, TextVerbosity: true, ExecutionControls: true, SubagentControl: true, ToolObservations: true, Preparation: true, LocalEnvironment: true, LocalEnvironmentNetworkPolicy: true, WorkspaceReadPreparation: true, WorkspaceOutputExport: true} + return proto.AgentKindCapabilities{Streaming: true, Steering: true, DurableTurns: true, DurableInputReceipts: true, EnvironmentNone: true, WebSearchControl: true, TextVerbosity: true, ExecutionControls: true, SubagentControl: true, ToolObservations: true, Preparation: true, LocalEnvironment: true, WorkspaceReadPreparation: true, WorkspaceOutputExport: true} } func workerEnvironmentReservation(t *testing.T, h *dispatchHarness) store.EnvironmentInputReservation { diff --git a/services/agents-api/internal/store/local_environment_worker_test.go b/services/agents-api/internal/store/local_environment_worker_test.go index 948a8fc95..b5dfc3381 100644 --- a/services/agents-api/internal/store/local_environment_worker_test.go +++ b/services/agents-api/internal/store/local_environment_worker_test.go @@ -26,7 +26,6 @@ func localWorker(t *testing.T, scoped, execute bool) (*dispatchHarness, *executi } caps := proto.AgentKindCapabilities{LocalEnvironment: true, Preparation: true, WorkspaceReadPreparation: true} if execute { - caps.LocalEnvironmentNetworkPolicy = true caps.WorkspaceOutputExport = true caps.Streaming, caps.Steering, caps.DurableTurns, caps.DurableInputReceipts = true, true, true, true caps.WebSearchControl, caps.TextVerbosity, caps.ExecutionControls = true, true, true diff --git a/services/agents-api/internal/store/runtime_node_lifecycle_test.go b/services/agents-api/internal/store/runtime_node_lifecycle_test.go index 3ba6bdfa6..1fc086b7f 100644 --- a/services/agents-api/internal/store/runtime_node_lifecycle_test.go +++ b/services/agents-api/internal/store/runtime_node_lifecycle_test.go @@ -54,10 +54,12 @@ func TestManagedNodesIsolateBlockedProviderAndInitialization(t *testing.T) { // Same-node direct callers and the scan all consume the same one-shot // allocation receipt. A's provider or Runtime operation remains blocked. - tenant, _, env := f.session(f.nodeB, true) + // The background scan can provision as soon as the Session is saved. + // Capture the baseline before making the new Environment visible. f.provider.mu.Lock() before := f.provider.creates f.provider.mu.Unlock() + tenant, _, env := f.session(f.nodeB, true) var callers sync.WaitGroup results := make(chan store.RuntimeAllocation, 12) failures := make(chan error, 12)