Address review: validate acceptance, sync handlers, spawn input type

- All three clients hard-fail registration when the reply omits a well-formed
  accepted_capabilities list (a missing/corrupt list means acceptance can't be
  confirmed); a malformed acceptance entry counts as rejected (fail-safe). Swift
  previously dropped malformed entries and could register on a corrupt reply.
- Python handlers may be sync or async (awaited only when awaitable), matching
  the TypeScript client.
- Swift spawnAgent takes [String: JSONValue] so callers can't build a node.spawn
  frame the engine's object-typed input schema rejects.
- Strengthen the reconnect test to complete re-registration after the drop.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: willwashburn

packages/sdk-python/src/relay_sdk/node.py

@@ -14,6 +14,7 @@
 
 import asyncio
 import contextlib
+import inspect
 import json
 import logging
 import random
@@ -130,7 +131,8 @@ class _RegisteredCapability:
     handler: "NodeCapabilityHandler"
 
 
-NodeCapabilityHandler = Callable[[Any, "NodeHandlerContext"], Awaitable[Any]]
+# A handler may be async (returning an awaitable) or a plain sync callable.
+NodeCapabilityHandler = Callable[[Any, "NodeHandlerContext"], Awaitable[Any] | Any]
 NodeCapabilitySpec = NodeCapabilityHandler | Mapping[str, Any]
 
 
@@ -469,11 +471,21 @@ async def _send_register(self) -> None:
             frame["machine_id"] = self._machine_id
 
         data = await self._request(req_id, frame)
-        accepted = (data or {}).get("accepted_capabilities") or []
-        rejected = [c for c in accepted if not c.get("accepted")]
+        # The reply must carry a well-formed acceptance list (the engine always
+        # does). A missing/corrupt list means acceptance can't be confirmed, so
+        # fail registration rather than assume success; an entry missing
+        # `accepted` counts as rejected (fail-safe).
+        accepted = (data or {}).get("accepted_capabilities")
+        if not isinstance(accepted, list):
+            raise NodeRegistrationError(
+                "Registration reply is missing accepted_capabilities", "invalid_register_reply"
+            )
+        rejected = [c for c in accepted if not (isinstance(c, dict) and c.get("accepted") is True)]
         if rejected:
             detail = ", ".join(
-                f"{c.get('name')}" + (f" ({c['reason']})" if c.get("reason") else "") for c in rejected
+                f"{c.get('name') if isinstance(c, dict) else '<unknown>'}"
+                + (f" ({c['reason']})" if isinstance(c, dict) and c.get("reason") else "")
+                for c in rejected
             )
             raise NodeRegistrationError(f"Capabilities rejected: {detail}", "capabilities_rejected")
 
@@ -540,7 +552,9 @@ async def _dispatch_invoke(self, invocation_id: str, action: str, input_: Any) -
             return
         ctx = NodeHandlerContext(self, invocation_id)
         try:
-            output = await cap.handler(input_, ctx)
+            # Support both async and plain (sync) handlers.
+            result = cap.handler(input_, ctx)
+            output = await result if inspect.isawaitable(result) else result
         except Exception as exc:  # a handler throw becomes an error result; never dropped
             await self._send_frame(
                 {"type": "action.result", "invocation_id": invocation_id, "error": str(exc)}

packages/sdk-python/tests/test_node.py

@@ -230,6 +230,45 @@ async def handler(input, ctx):
     await asyncio.wait_for(task, timeout=1.0)
 
 
+@pytest.mark.asyncio
+async def test_invoke_supports_a_sync_handler():
+    server = FakeNodeServer()
+    node = NodeProvider(**base_kwargs(server))
+
+    @node.capability("run-etl")
+    def handler(input, ctx):  # a plain (non-async) handler
+        return {"echoed": input}
+
+    task = asyncio.create_task(node.serve())
+    conn = await server.next_connection()
+    conn.push(accept_all(conn.last_register()))
+    await node.wait_registered()
+
+    conn.push({"v": 1, "type": "action.invoke", "invocation_id": "inv-sync", "action": "run-etl", "input": {"rows": 2}})
+    await wait_until(lambda: len(conn.sent_of_type("action.result")) == 1)
+    result = conn.sent_of_type("action.result")[-1]
+    assert result["output"] == {"echoed": {"rows": 2}}
+
+    await node.stop()
+    await asyncio.wait_for(task, timeout=1.0)
+
+
+@pytest.mark.asyncio
+async def test_registration_missing_accepted_capabilities_raises():
+    server = FakeNodeServer()
+    node = NodeProvider(**base_kwargs(server))
+
+    task = asyncio.create_task(node.serve())
+    conn = await server.next_connection()
+    register = conn.last_register()
+    # A reply that confirms nothing about capability acceptance is not success.
+    conn.push({"v": 1, "id": register["id"], "type": "reply", "ok": True, "data": {"provider": register["provider"]}})
+    with pytest.raises(NodeRegistrationError):
+        await asyncio.wait_for(node.wait_registered(), timeout=1.0)
+    with pytest.raises(NodeRegistrationError):
+        await asyncio.wait_for(task, timeout=1.0)
+
+
 @pytest.mark.asyncio
 async def test_handler_throw_becomes_error_result_never_dropped():
     server = FakeNodeServer()

packages/sdk-swift/Sources/Relaycast/NodeProvider.swift

@@ -54,13 +54,13 @@ public struct NodeHandlerContext: Sendable {
     public let invocationID: String
 
     private let sendMessageImpl: @Sendable (String, String, String, NodeMessageMode?, [String: JSONValue]?) async throws -> JSONValue
-    private let spawnAgentImpl: @Sendable (JSONValue) async throws -> JSONValue
+    private let spawnAgentImpl: @Sendable ([String: JSONValue]) async throws -> JSONValue
 
     init(
         node: NodeInfo,
         invocationID: String,
         sendMessageImpl: @escaping @Sendable (String, String, String, NodeMessageMode?, [String: JSONValue]?) async throws -> JSONValue,
-        spawnAgentImpl: @escaping @Sendable (JSONValue) async throws -> JSONValue
+        spawnAgentImpl: @escaping @Sendable ([String: JSONValue]) async throws -> JSONValue
     ) {
         self.node = node
         self.invocationID = invocationID
@@ -83,9 +83,10 @@ public struct NodeHandlerContext: Sendable {
 
     /// Spawn an agent through the node's capacity executor. Capacity-direct: it
     /// never re-enters action dispatch, so a `spawn:<harness>` shadow handler
-    /// that delegates cannot recurse into itself. Resolves with the spawn
-    /// placement.
-    public func spawnAgent(_ input: JSONValue) async throws -> JSONValue {
+    /// that delegates cannot recurse into itself. The spawn spec is a JSON
+    /// object (the engine's `node.spawn` input is object-typed). Resolves with
+    /// the spawn placement.
+    public func spawnAgent(_ input: [String: JSONValue]) async throws -> JSONValue {
         try await spawnAgentImpl(input)
     }
 }
@@ -109,25 +110,33 @@ public struct NodeRegisterReplyData: Equatable, Sendable {
     public let raw: JSONValue
 
     static func parse(_ json: JSONValue) throws -> NodeRegisterReplyData {
+        // A malformed reply is a deterministic registration failure (thrown as a
+        // NodeRegistrationError so it hard-fails rather than being retried as a
+        // transient drop).
         guard case .object(let obj) = json,
               case .object(let providerObj)? = obj["provider"],
               case .string(let name)? = providerObj["name"],
               case .string(let instanceID)? = providerObj["instance_id"]
         else {
-            throw NodeProviderError.invalidRegisterReply
+            throw NodeRegistrationError(code: "invalid_register_reply", message: "node.register reply is malformed")
         }
 
-        var acceptedCapabilities: [NodeCapabilityAcceptance] = []
-        if case .array(let items)? = obj["accepted_capabilities"] {
-            acceptedCapabilities = items.compactMap { item -> NodeCapabilityAcceptance? in
-                guard case .object(let capObj) = item,
-                      case .string(let capName)? = capObj["name"],
-                      case .bool(let accepted)? = capObj["accepted"]
-                else { return nil }
-                let kind: String = { if case .string(let k)? = capObj["kind"] { return k } else { return "action" } }()
-                let reason: String? = { if case .string(let r)? = capObj["reason"] { return r } else { return nil } }()
-                return NodeCapabilityAcceptance(name: capName, kind: kind, accepted: accepted, reason: reason)
+        // The reply must carry a well-formed acceptance list (the engine always
+        // does). A missing/corrupt list means acceptance can't be confirmed, so
+        // fail registration rather than assume success.
+        guard case .array(let items)? = obj["accepted_capabilities"] else {
+            throw NodeRegistrationError(code: "invalid_register_reply", message: "node.register reply is missing accepted_capabilities")
+        }
+        let acceptedCapabilities: [NodeCapabilityAcceptance] = items.map { item -> NodeCapabilityAcceptance in
+            // A malformed entry can't be confirmed accepted — count it as
+            // rejected (fail-safe) rather than dropping it.
+            guard case .object(let capObj) = item, case .string(let capName)? = capObj["name"] else {
+                return NodeCapabilityAcceptance(name: "<unknown>", kind: "action", accepted: false, reason: "malformed acceptance entry")
             }
+            let accepted: Bool = { if case .bool(let a)? = capObj["accepted"] { return a } else { return false } }()
+            let kind: String = { if case .string(let k)? = capObj["kind"] { return k } else { return "action" } }()
+            let reason: String? = { if case .string(let r)? = capObj["reason"] { return r } else { return nil } }()
+            return NodeCapabilityAcceptance(name: capName, kind: kind, accepted: accepted, reason: reason)
         }
 
         return NodeRegisterReplyData(providerName: name, instanceID: instanceID, acceptedCapabilities: acceptedCapabilities, raw: json)
@@ -682,12 +691,12 @@ public actor NodeProvider {
         return try await request(payload)
     }
 
-    private func spawnAgentFrame(_ input: JSONValue) async throws -> JSONValue {
+    private func spawnAgentFrame(_ input: [String: JSONValue]) async throws -> JSONValue {
         // Capacity-direct: the engine always targets this connection's own node,
         // so the frame carries no node target.
         try await request([
             "type": .string("node.spawn"),
-            "input": input
+            "input": .object(input)
         ])
     }
 

packages/sdk-swift/Tests/RelaycastTests/NodeProviderTests.swift

@@ -216,6 +216,47 @@ final class NodeProviderTests: XCTestCase {
         }
     }
 
+    func testHardFailsRegistrationWhenReplyOmitsAcceptedCapabilities() async throws {
+        let factory = FakeTransportFactory()
+        let node = NodeProvider(
+            nodeToken: baseOptions.nodeToken,
+            nodeID: baseOptions.nodeID,
+            nodeName: baseOptions.nodeName,
+            provider: (name: "py", instanceID: nil),
+            transport: { factory.make() }
+        )
+        await node.capability("run-etl") { input, _ in input }
+
+        let serveTask = Task { try await node.serve() }
+
+        await waitUntil { factory.instances.count == 1 }
+        let transport = factory.instances[0]
+        await waitUntil { await !transport.sentOfType("node.register").isEmpty }
+        let register = await transport.lastRegister()!
+
+        // A reply that confirms nothing about capability acceptance is not success.
+        await transport.emit([
+            "id": register["id"] ?? .null,
+            "type": .string("reply"),
+            "ok": .bool(true),
+            "data": .object(["provider": register["provider"] ?? .object([:])])
+        ])
+
+        do {
+            _ = try await node.waitRegistered()
+            XCTFail("expected waitRegistered() to throw")
+        } catch let error as NodeRegistrationError {
+            XCTAssertEqual(error.code, "invalid_register_reply")
+        }
+
+        do {
+            try await serveTask.value
+            XCTFail("expected serve() to throw")
+        } catch {
+            // expected
+        }
+    }
+
     func testHardFailsRegistrationOnEngineErrorFrame() async throws {
         let factory = FakeTransportFactory()
         let node = NodeProvider(
@@ -450,7 +491,7 @@ final class NodeProviderTests: XCTestCase {
             transport: { factory.make() }
         )
         await node.capability("spawn:claude") { _, ctx in
-            try await ctx.spawnAgent(.object(["cli": .string("claude"), "name": .string("worker-1")]))
+            try await ctx.spawnAgent(["cli": .string("claude"), "name": .string("worker-1")])
         }
 
         let serveTask = Task { try await node.serve() }

packages/sdk-typescript/src/__tests__/node-provider.test.ts

@@ -238,6 +238,23 @@ describe('NodeProviderClient', () => {
     const secondInstance = (secondRegister.provider as { instance_id: string }).instance_id;
     expect(secondRegister).toMatchObject({ name: 'alpha', provider: { name: 'py' } });
     expect(secondInstance).not.toBe(firstInstance);
+
+    // The reconnected registration completes its lifecycle: accepting it brings
+    // the provider back to connected.
+    second.emit(acceptAll(secondRegister));
+    await vi.waitFor(() => expect(node.connected).toBe(true));
+  });
+
+  it('hard-fails registration when the reply omits accepted_capabilities', async () => {
+    const node = new NodeProviderClient({ ...baseOptions, capabilities: { 'run-etl': async () => 'ok' } });
+    node.serve().catch(() => {});
+    const sock = newSocket();
+    sock.open();
+    const register = sock.lastRegister();
+    // A reply that confirms nothing about capability acceptance must not be
+    // treated as success.
+    sock.emit({ v: 1, id: register.id, type: 'reply', ok: true, data: { provider: register.provider } });
+    await expect(node.whenRegistered()).rejects.toMatchObject({ code: 'invalid_register_reply' });
   });
 
   it('reconnects after a transport drop during the register handshake', async () => {

packages/sdk-typescript/src/node-provider.ts

@@ -350,9 +350,21 @@ export class NodeProviderClient {
   }
 
   private onRegistered(data: NodeRegisterReplyData): void {
-    const rejected = (data.accepted_capabilities ?? []).filter((c: FleetCapabilityAcceptance) => !c.accepted);
+    // The reply must carry a well-formed acceptance list (the engine always
+    // does). A missing/corrupt list means acceptance can't be confirmed, so
+    // fail registration rather than assume success; an entry missing `accepted`
+    // counts as rejected (fail-safe).
+    const list = data.accepted_capabilities;
+    if (!Array.isArray(list)) {
+      this.onRegisterFailed(new NodeRegistrationError(
+        'Registration reply is missing accepted_capabilities',
+        'invalid_register_reply',
+      ));
+      return;
+    }
+    const rejected = list.filter((c: FleetCapabilityAcceptance) => !c || c.accepted !== true);
     if (rejected.length > 0) {
-      const detail = rejected.map((c) => `${c.name}${c.reason ? ` (${c.reason})` : ''}`).join(', ');
+      const detail = rejected.map((c) => `${c?.name ?? '<unknown>'}${c?.reason ? ` (${c.reason})` : ''}`).join(', ');
       this.onRegisterFailed(new NodeRegistrationError(`Capabilities rejected: ${detail}`, 'capabilities_rejected'));
       return;
     }