fix: harden token-only runtime workers

CHANGELOG.md

@@ -18,6 +18,12 @@ This is a pre-1.0 breaking Runtime cutover.
   protocol generation.
 - Documented Agent Node as an optional migration adapter rather than the default
   Runtime path.
+- Token-only Workers now derive a deterministic, token-scoped Node ID when
+  `node_id` is omitted; explicit mTLS deployments still require their
+  provisioned Node identity.
+- Classified only enumerated permanent HTTP error and WebSocket close shapes as
+  fatal. Unknown auth-like failures remain recoverable for DB-backed
+  revalidation.
 - Added complete bilingual repository, contribution, security, support, release,
   package metadata, typing marker, and MIT license files for the first public
   Python SDK release.

README.md

@@ -214,7 +214,7 @@ asyncio.run(main())
 
 ## Runtime Worker
 
-`RuntimeWorker` owns Runtime discovery, mTLS, Session attachment, WebSocket-first
+`RuntimeWorker` owns Runtime discovery, token-only or mTLS security, Session attachment, WebSocket-first
 delivery with HTTPS long-poll fallback, assignment confirmation, lease renewal,
 resume, cancellation, drain, and shutdown. The handler starts only after Core
 confirms the assignment.
@@ -235,6 +235,7 @@ async def handle(context: runtime.RuntimeContext) -> runtime.RuntimeResult:
 async def main() -> None:
     worker = runtime.RuntimeWorker(
         platform_url=os.environ["OPENLINKER_URL"],
+        agent_id=os.environ["OPENLINKER_AGENT_ID"],
         agent_token=os.environ["OPENLINKER_AGENT_TOKEN"],
         data_dir=os.environ.get(
             "OPENLINKER_RUNTIME_DATA_DIR",
@@ -253,7 +254,9 @@ A worker is async and single-use. Events and results are encrypted and fsynced
 before upload. Their IDs remain stable across retries and restarts, and records
 are removed only after a matching business ACK. The file store keeps a stable
 worker identity while rotating the Session identity for each process start.
-The SDK also generates the Node private key in `data_dir`, enrolls it with the
+For token-only security, omitting `node_id` derives the same stable
+token-scoped Node ID as the Go and TypeScript SDKs. When discovery requires
+mTLS, the SDK generates the Node private key in `data_dir`, enrolls it with the
 Agent Token, and renews its 24-hour client certificate automatically. Explicit
 `RuntimeMTLS` file paths are only needed for external-PKI compatibility.
 

README.zh-CN.md

@@ -205,7 +205,7 @@ asyncio.run(main())
 
 ## Runtime Worker
 
-`RuntimeWorker` 负责 Runtime 发现、mTLS、Session attach、WebSocket 主链路与 HTTPS
+`RuntimeWorker` 负责 Runtime 发现、token-only/mTLS 安全策略、Session attach、WebSocket 主链路与 HTTPS
 长轮询回退、任务确认、租约续期、恢复、取消、drain 和关闭。只有 Core 确认任务后才会
 执行 Handler。
 
@@ -225,15 +225,8 @@ async def handle(context: runtime.RuntimeContext) -> runtime.RuntimeResult:
 async def main() -> None:
     worker = runtime.RuntimeWorker(
         platform_url=os.environ["OPENLINKER_URL"],
-        runtime_url=os.environ.get("OPENLINKER_RUNTIME_URL", ""),
-        node_id=os.environ["OPENLINKER_NODE_ID"],
         agent_id=os.environ["OPENLINKER_AGENT_ID"],
         agent_token=os.environ["OPENLINKER_AGENT_TOKEN"],
-        mtls=runtime.RuntimeMTLS(
-            cert_file=os.environ["OPENLINKER_RUNTIME_MTLS_CERT_FILE"],
-            key_file=os.environ["OPENLINKER_RUNTIME_MTLS_KEY_FILE"],
-            ca_file=os.environ["OPENLINKER_RUNTIME_MTLS_CA_FILE"],
-        ),
         data_dir=os.environ.get(
             "OPENLINKER_RUNTIME_DATA_DIR",
             "./.openlinker-runtime",
@@ -247,6 +240,10 @@ async def main() -> None:
 asyncio.run(main())
 ```
 
+发现选择 token-only 且省略 `node_id` 时，SDK 会派生与 Go/TypeScript SDK 一致、仅限该
+Agent Token 的稳定 Node ID。发现要求 mTLS 时由 SDK 自动登记；只有外部 PKI 兼容模式才需
+显式证书文件。
+
 Worker 是异步且一次性的。Event 和 Result 会在上传前加密并 fsync；重试和重启不会改变
 它们的 ID，只有取得匹配的业务 ACK 后才会删除记录。文件 Store 会保留稳定的 worker
 identity，并在每次进程启动时更换 Session identity。

examples/runtime_echo.py

@@ -33,14 +33,14 @@ async def main() -> None:
     worker = runtime.RuntimeWorker(
         platform_url=os.environ["OPENLINKER_URL"],
         runtime_url=os.environ.get("OPENLINKER_RUNTIME_URL", ""),
-        node_id=os.environ["OPENLINKER_NODE_ID"],
+        node_id=os.environ.get("OPENLINKER_NODE_ID", ""),
         node_version="openlinker-python/runtime-worker",
         agent_id=os.environ["OPENLINKER_AGENT_ID"],
         agent_token=os.environ["OPENLINKER_AGENT_TOKEN"],
         mtls=runtime.RuntimeMTLS(
-            cert_file=os.environ["OPENLINKER_RUNTIME_MTLS_CERT_FILE"],
-            key_file=os.environ["OPENLINKER_RUNTIME_MTLS_KEY_FILE"],
-            ca_file=os.environ["OPENLINKER_RUNTIME_MTLS_CA_FILE"],
+            cert_file=os.environ.get("OPENLINKER_RUNTIME_MTLS_CERT_FILE", ""),
+            key_file=os.environ.get("OPENLINKER_RUNTIME_MTLS_KEY_FILE", ""),
+            ca_file=os.environ.get("OPENLINKER_RUNTIME_MTLS_CA_FILE", ""),
         ),
         data_dir=os.environ.get("OPENLINKER_RUNTIME_DATA_DIR", "./.openlinker-runtime"),
         transport=os.environ.get("OPENLINKER_RUNTIME_TRANSPORT", "auto"),

src/openlinker/runtime/transport.py

@@ -3,6 +3,7 @@
 import asyncio
 import inspect
 import json
+import re
 import ssl
 import uuid
 from dataclasses import dataclass
@@ -39,38 +40,7 @@
 RUNTIME_FALLBACK_REASONS = frozenset(
     {"explicit", "websocket_unavailable", "policy_forced", "recovery"}
 )
-_ERROR_CODES = {
-    "BAD_REQUEST",
-    "UNAUTHORIZED",
-    "FORBIDDEN",
-    "PERMISSION_DENIED",
-    "NOT_FOUND",
-    "CONFLICT",
-    "VALIDATION_FAILED",
-    "RATE_LIMITED",
-    "INTERNAL_ERROR",
-    "SERVICE_UNAVAILABLE",
-    "IDEMPOTENCY_KEY_REUSED",
-    "RUN_ALREADY_TERMINAL",
-    "STALE_LEASE",
-    "LEASE_EXPIRED",
-    "LEASE_IDENTITY_MISMATCH",
-    "RESULT_ID_CONFLICT",
-    "EVENT_ID_CONFLICT",
-    "NODE_AT_CAPACITY",
-    "RUNTIME_CLIENT_UPGRADE_REQUIRED",
-    "RUNTIME_REQUIRED_FEATURE_MISSING",
-    "RUN_CANCEL_REQUESTED",
-    "RUN_CANCEL_UNCONFIRMED",
-    "RUNTIME_RETRY_EXHAUSTED",
-    "RUNTIME_DISPATCH_TIMEOUT",
-    "RUN_DEADLINE_EXCEEDED",
-    "EVENTS_MISSING",
-    "REPLAY_INPUT_UNAVAILABLE",
-    "ENDPOINT_RESULT_UNKNOWN",
-    "RUNTIME_SESSION_CONFLICT",
-    "RUNTIME_SPOOL_CORRUPT",
-}
+_ERROR_CODE = re.compile(r"^[A-Z][A-Z0-9_]{0,119}$")
 
 
 @dataclass(frozen=True)
@@ -1129,8 +1099,11 @@ def _parse_error_body(value: dict[str, Any], *, status_code: int = 0) -> Runtime
     code = value["code"]
     message = value["message"]
     retryable = value.get("retryable", False)
-    if not isinstance(code, str) or code not in _ERROR_CODES:
-        raise RuntimeProtocolError("Runtime returned an unknown error code")
+    # Error bodies remain structurally strict, but the code enumeration is
+    # deliberately forward-compatible. Worker fatality is a best-effort
+    # classifier: only explicitly known permanent codes may terminate it.
+    if not isinstance(code, str) or _ERROR_CODE.fullmatch(code) is None:
+        raise RuntimeProtocolError("Runtime returned an invalid error code")
     if not isinstance(message, str) or not 1 <= len(message) <= 500:
         raise RuntimeProtocolError("Runtime error message is invalid")
     if not isinstance(retryable, bool):

src/openlinker/runtime/worker.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import hashlib
 import inspect
 import logging
 import random
@@ -78,6 +79,7 @@
 DEFAULT_NODE_VERSION = "openlinker-python/runtime-worker"
 
 _PERMANENT_CODES = {
+    "AUTHENTICATION_FAILED",
     "UNAUTHORIZED",
     "FORBIDDEN",
     "PERMISSION_DENIED",
@@ -86,6 +88,7 @@
     "RUNTIME_SESSION_CONFLICT",
     "RUNTIME_SPOOL_CORRUPT",
 }
+_PERMANENT_WEBSOCKET_CLOSE_CODES = {4401, 4406, 4409, 4412}
 _LEASE_TERMINAL_CODES = {"STALE_LEASE", "LEASE_EXPIRED", "RUN_ALREADY_TERMINAL"}
 _ASSIGNMENT_TERMINAL_CODES = _LEASE_TERMINAL_CODES | {"RUN_CANCEL_REQUESTED"}
 
@@ -549,10 +552,12 @@ async def _setup_transport(self) -> None:
             origin, allow_loopback_http=not self._mtls_required
         )
         if not self._mtls_required:
-            if not self.node_id or not self.agent_id:
+            if not self.agent_id:
                 raise ValueError(
-                    "node_id and agent_id are required for token-only Runtime transport"
+                    "agent_id is required for token-only Runtime transport"
                 )
+            if not self.node_id:
+                self.node_id = _token_scoped_runtime_node_id(self.agent_token)
         elif not explicit_mtls:
             if self.data_dir is None:
                 raise ValueError("automatic Runtime credentials require data_dir")
@@ -2102,7 +2107,8 @@ async def _retry_call(
                         await self._wait_or_stop(self.retry_minimum)
                     continue
                 if _fatal_error(exc) or (
-                    isinstance(exc, RuntimeRemoteError) and exc.code in _LEASE_TERMINAL_CODES
+                    isinstance(exc, RuntimeRemoteError)
+                    and exc.code in _ASSIGNMENT_TERMINAL_CODES
                 ):
                     raise
                 if deadline is not None and datetime.now(timezone.utc) >= deadline:
@@ -2425,9 +2431,13 @@ def _fatal_error(exc: BaseException) -> bool:
         return True
     if isinstance(exc, (RuntimeProtocolError, RuntimeStoreError)):
         return True
-    return isinstance(exc, RuntimeRemoteError) and (
-        exc.code in _PERMANENT_CODES or (not exc.retryable and 400 <= exc.status_code < 500)
-    )
+    if isinstance(exc, RuntimeRemoteError):
+        return exc.code in _PERMANENT_CODES
+    if isinstance(exc, ConnectionClosed):
+        received = getattr(exc, "rcvd", None)
+        code = getattr(received, "code", None) if received is not None else getattr(exc, "code", None)
+        return code in _PERMANENT_WEBSOCKET_CLOSE_CODES
+    return False
 
 
 def _session_conflict(exc: BaseException) -> bool:
@@ -2614,6 +2624,17 @@ def _canonical_uuid(value: str, label: str) -> None:
         raise ValueError(f"{label} must be a lowercase non-zero UUID")
 
 
+def _token_scoped_runtime_node_id(agent_token: str) -> str:
+    digest = hashlib.sha256(
+        b"openlinker/runtime-worker/token-scoped-node/v1\x00"
+        + agent_token.encode("utf-8")
+    ).digest()
+    value = bytearray(digest[:16])
+    value[6] = (value[6] & 0x0F) | 0x50
+    value[8] = (value[8] & 0x3F) | 0x80
+    return str(uuid.UUID(bytes=bytes(value)))
+
+
 def _canonical_protocol_uuid(value: str, label: str) -> None:
     try:
         _canonical_uuid(value, label)

tests/test_runtime.py

@@ -67,6 +67,25 @@ def assignment(store: runtime.MemoryRuntimeStore) -> runtime.RuntimeAssignment:
     )
 
 
+def test_runtime_error_classifier_is_explicit_and_forward_compatible():
+    assert runtime_worker_module._fatal_error(
+        runtime.RuntimeRemoteError("UNAUTHORIZED", "revoked", status_code=401)
+    )
+    assert runtime_worker_module._fatal_error(
+        ConnectionClosedError(Close(4401, "AUTHENTICATION_FAILED"), None)
+    )
+    assert not runtime_worker_module._fatal_error(
+        runtime.RuntimeRemoteError(
+            "AUTH_BACKEND_BUSY",
+            "temporary authentication backend failure",
+            status_code=401,
+        )
+    )
+    assert not runtime_worker_module._fatal_error(
+        ConnectionClosedError(Close(4499, "AUTH_BACKEND_BUSY"), None)
+    )
+
+
 class FakeTransport:
     def __init__(self, *, kind: str = "pull") -> None:
         self.kind = kind
@@ -388,7 +407,9 @@ def build_http(*_args, **kwargs):
 
 
 @pytest.mark.asyncio
-async def test_worker_token_only_mode_requires_configured_identity(monkeypatch):
+async def test_worker_token_only_mode_derives_node_identity(monkeypatch):
+    captured: dict[str, Any] = {}
+
     async def discover(_platform_url):
         return RuntimeDiscoveryConnection(
             "https://runtime.example.test",
@@ -397,17 +418,27 @@ async def discover(_platform_url):
         )
 
     monkeypatch.setattr(runtime_worker_module, "discover_runtime_connection", discover)
+    monkeypatch.setattr(
+        runtime_worker_module,
+        "HTTPRuntimeTransport",
+        lambda *_args, **kwargs: captured.update(kwargs) or FakeTransport(),
+    )
     worker = runtime.RuntimeWorker(
         platform_url="https://platform.example.test",
+        agent_id=AGENT_ID,
         agent_token="ol_agent_token_only",
         store=runtime.MemoryRuntimeStore(),
         allow_unsafe_memory_store=True,
         handler=lambda _context: {},
         transport="pull",
     )
 
-    with pytest.raises(ValueError, match="required for token-only"):
-        await worker._setup_transport()
+    await worker._setup_transport()
+    assert worker.node_id == runtime_worker_module._token_scoped_runtime_node_id(
+        "ol_agent_token_only"
+    )
+    assert worker.node_id == "d6bb911d-7ad6-528b-9a8e-34e2785975fd"
+    assert captured["node_id"] == worker.node_id
 
 
 @pytest.mark.asyncio

tests/test_runtime_transport.py

@@ -35,6 +35,27 @@
 NODE_ID = "11111111-1111-4111-8111-111111111111"
 
 
+def test_runtime_error_body_accepts_future_well_formed_codes():
+    error = runtime_transport._parse_error_body(
+        {
+            "code": "AUTH_BACKEND_BUSY",
+            "message": "temporary authentication backend failure",
+        },
+        status_code=401,
+    )
+    assert isinstance(error, runtime.RuntimeRemoteError)
+    assert error.code == "AUTH_BACKEND_BUSY"
+    assert error.status_code == 401
+
+
+def test_runtime_error_body_still_rejects_malformed_codes():
+    with pytest.raises(runtime.RuntimeProtocolError, match="invalid error code"):
+        runtime_transport._parse_error_body(
+            {"code": "auth backend busy", "message": "invalid"},
+            status_code=401,
+        )
+
+
 def test_runtime_discovery_policy_fixtures_are_language_consistent():
     fixture = json.loads(
         (Path(__file__).parents[1] / "contracts/runtime-discovery-policy-fixtures.json").read_text()