fix(harness): limit provider credential routes to workload commands

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

docs/harness-contract.md

@@ -80,9 +80,17 @@ and its dependencies available inside that transport; host-only installation is
 not treated as sandbox availability.
 
 The gateway binds local execution to the exact create-time consumer: actor,
-project, and resolved request id as `run_id`. Every owner command forwards the
+project, and resolved request id as `run_id`. Housekeeping `run` commands
-selected profile's value-free `credential_route_refs` and a timeout capped by
+(executable lookup, task creation/removal, Git inspection and artifact capture)
-its limit. The binding is process-local and excluded from serialized handles.
+request no credential routes. Only the selected rein's explicit `run_workload`
+command forwards the selected profile's value-free `credential_route_refs`.
+Both bindings cap timeouts by the profile limit and remain process-local,
+excluded from serialized handles. An absent workload binding refuses rather
+than using the housekeeping binding. SSH profiles with credential routes refuse
+because SSH has no owner-mediated credential delivery contract. These bindings
+narrow trusted gateway behavior; they do not authenticate arbitrary Python
+callers, remove task-scoped MessagesOwner routes or establish isolation of
+the caller's interactive runtime.
 An unbound local descriptor refuses execution; no consumer namespace entry or
 host execution fallback exists. Owner timeouts and truncated output fail closed.
 Generated task specs travel over stdin to an exclusive mode-0600 writer inside
@@ -187,3 +195,10 @@ model pin is a requested route, and does not fill absent observed metadata.
 Unknown remains null, including failure before provider execution. A reported
 alias does not attest an exact provider revision or conversation identifier.
 See `docs/evidence/2026-10-03-emergent-session-readiness-review.md`.
+
+
+`scripts/prove-credential-command-scope.py` exercises the Glas transport against
+a disposable real bwrap namespace with synthetic material. It checks host-file
+absence, inherited credential-variable denial through a subprocess, explicit
+workload-only provider delivery, output redaction, subsequent-command absence
+and teardown. It uses no provider or model and cannot promote a runtime profile.

scripts/prove-credential-command-scope.py

@@ -0,0 +1,55 @@
+"""No model/provider: actual bwrap transport proof with synthetic credentials."""
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+from sandboxer.extensions.bwrap import BwrapExtension
+from sandboxer.models import Profile
+from glas_harness.contract import SandboxHandle
+from glas_harness.transport import transport_from_sandbox
+
+
+def main():
+    with tempfile.TemporaryDirectory(prefix="glas-command-scope-") as tmp:
+        root = Path(tmp)
+        sentinel = root / "admin-sentinel"
+        sentinel.write_text("synthetic-file-only")
+        provider = root / "provider.py"
+        provider.write_text("import os,subprocess,sys\nenv=os.environ.copy()\nenv['ANTHROPIC_API_KEY']='synthetic-provider-only'\nraise SystemExit(subprocess.call(sys.argv[sys.argv.index('--')+1:],env=env))\n")
+        ext = BwrapExtension({"base_dir": str(root / "sandboxes"), "credential_routes": {
+            "synthetic": {"profiles":["profile.scope-proof"], "projects":["scope-proof"], "actors":["agt"], "exec_argv":[sys.executable,str(provider),"--"]}}})
+        context = {"actor":"agt", "project":"scope-proof", "run_id":"scope-proof-1", "profile_id":"profile.scope-proof"}
+        # Inherited values are synthetic. Real owner credentials are never read.
+        before = {key:os.environ.get(key) for key in ("BAO_TOKEN","VAULT_TOKEN","AWS_SECRET_ACCESS_KEY","SSH_AUTH_SOCK")}
+        for key in before:
+            os.environ[key] = "synthetic-inherited-only"
+        handle = None
+        try:
+            handle = ext.provision(Profile(id="profile.scope-proof",version="1",extension="ext.bwrap"), {}, "localhost")
+            ext.wait_ready(handle)
+            sandbox = SandboxHandle(sandbox_id=handle["sandbox_id"],host="localhost",reachability={"pid":handle["pid"],"workspace_dir":handle["workspace_dir"]})
+            calls=[]
+            def execute(argv, input_text, timeout, routes=()):
+                calls.append(list(routes))
+                raw=ext.execute(handle,argv,credential_route_refs=list(routes),execution_context=context,timeout_seconds=timeout,max_output_bytes=4096,stdin_text=input_text)
+                return type("Result",(),raw)()
+            sandbox._owner_execute=execute
+            sandbox._owner_workload_execute=lambda argv,stdin,timeout: execute(argv,stdin,timeout,["synthetic"])
+            transport=transport_from_sandbox(sandbox)
+            probe = "import os,pathlib,subprocess; assert not pathlib.Path("+repr(str(sentinel))+").exists(); assert all(k not in os.environ for k in "+repr(list(before))+"); assert subprocess.check_output(['python3','-c',\"import os; print(os.environ.get('BAO_TOKEN','absent'))\"]).strip()==b'absent'; "
+            for workload in (False,True,False):
+                check = probe + ("assert os.environ['ANTHROPIC_API_KEY']=='synthetic-provider-only'; print(os.environ['ANTHROPIC_API_KEY'])" if workload else "assert 'ANTHROPIC_API_KEY' not in os.environ; print('clean')")
+                result=(transport.run_workload if workload else transport.run)(["python3","-c",check],timeout=10)
+                assert result.returncode==0, result.stderr
+                assert result.stdout == ("[REDACTED]\n" if workload else "clean\n")
+            assert calls==[[],["synthetic"],[]]
+        finally:
+            teardown=ext.teardown(handle) if handle else {}
+            for key,value in before.items():
+                if value is None: os.environ.pop(key,None)
+                else: os.environ[key]=value
+        assert teardown.get("workspace_removed")=="True"
+        print(json.dumps({"scope":"disposable bwrap via Glas transport; not interactive assignment", "model_calls":0,"real_credentials_read":False,"host_file_denied":True,"inherited_env_and_subprocess_denied":True,"credential_routes_by_command":calls,"provider_output_redacted":True,"subsequent_command_clean":True,"workspace_removed":True}))
+
+if __name__ == "__main__": main()

src/glas_harness/contract.py

@@ -117,6 +117,8 @@ class ReinDescriptor(ContractModel):
 class SandboxHandle(ContractModel):
     # Process-local owner binding; never serialized as reachability or evidence.
     _owner_execute: Any = PrivateAttr(default=None)
+    _owner_workload_execute: Any = PrivateAttr(default=None)
+    _credential_route_refs: list[str] = PrivateAttr(default_factory=list)
     _timeout_seconds: int = PrivateAttr(default=600)
 
     sandbox_id: str

src/glas_harness/gateway.py

@@ -7,6 +7,7 @@ There is deliberately no default rein or model in governed execution.
 from __future__ import annotations
 
 import time
+from functools import partial
 from collections.abc import Callable
 import uuid
 from datetime import UTC, datetime
@@ -145,19 +146,23 @@ def run_execution(
                 reachability=reachability,
             )
 
-            def owner_execute(argv, input_text, timeout):
+            def owner_execute(argv, input_text, timeout, *, credential_routes=()):
                 return manager.execute(
                     sandbox.sandbox_id,
                     SandboxExecRequest(
                         command=list(argv),
                         consumer=consumer,
-                        credential_route_refs=profile.credential_route_refs,
+                        credential_route_refs=list(credential_routes),
                         stdin_text=input_text,
                         timeout_seconds=timeout,
                     ),
                 )
 
             sandbox._owner_execute = owner_execute
+            sandbox._owner_workload_execute = partial(
+                owner_execute, credential_routes=tuple(profile.credential_route_refs)
+            )
+            sandbox._credential_route_refs = list(profile.credential_route_refs)
             sandbox._timeout_seconds = profile.limits.timeout_seconds or 600
 
             session = selected_rein.start_session(

src/glas_harness/reins/rein_aharness.py

@@ -109,7 +109,7 @@ class ReinAharness(Rein):
             argv += ["--tool-profile", self.tool_profile]
         if self.budget_tokens:
             argv += ["--budget-tokens", str(self.budget_tokens)]
-        proc = transport.run(argv, timeout=session["timeout_seconds"])
+        proc = transport.run_workload(argv, timeout=session["timeout_seconds"])
         ok = proc.returncode == 0
         output, events = self._split_stream_events(proc.stdout)
         last_result = session["last_result"] = parse_json_object(output)

src/glas_harness/reins/rein_openweights.py

@@ -91,7 +91,7 @@ class ReinOpenWeights(Rein):
             argv += ["--budget-tokens", str(self.budget_tokens)]
         if self.tool_profile:
             argv += ["--tool-profile", self.tool_profile]
-        proc = transport.run(argv, timeout=session["timeout_seconds"])
+        proc = transport.run_workload(argv, timeout=session["timeout_seconds"])
         ok = proc.returncode == 0
         last_result = session["last_result"] = parse_json_object(proc.stdout)
         return ToolResult(

src/glas_harness/transport.py

@@ -13,7 +13,7 @@ import re
 import shlex
 import subprocess
 import uuid
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path, PurePosixPath
 from typing import Any, Callable, Literal, Sequence
 
@@ -36,6 +36,7 @@ class ExecutionTransport:
     pid: int | None = None
     ssh_target: str | None = None
     owner_execute: Callable | None = None
+    owner_workload_execute: Callable | None = None
     timeout_seconds: int = 600
 
     def command(self, argv: Sequence[str]) -> list[str]:
@@ -94,6 +95,26 @@ class ExecutionTransport:
             timeout=bounded_timeout,
         )
 
+    def run_workload(
+        self,
+        argv: Sequence[str],
+        *,
+        input_text: str | None = None,
+        timeout: float | None = None,
+    ) -> subprocess.CompletedProcess[str]:
+        """Run the selected rein; only this path requests provider routes.
+
+        The gateway is trusted control-plane code. This separation narrows its
+        credential delivery; it is not authentication of arbitrary Python callers.
+        """
+        if self.kind == "local_namespace":
+            if self.owner_workload_execute is None:
+                raise TransportError("workload requires an owner workload binding")
+            return replace(self, owner_execute=self.owner_workload_execute).run(
+                argv, input_text=input_text, timeout=timeout
+            )
+        return self.run(argv, input_text=input_text, timeout=timeout)
+
     def resolve_executable(self, name: str, *, timeout: float = 15.0) -> str:
         if not name or "/" in name:
             raise TransportError("rein executable must be a command name, not a host path")
@@ -169,9 +190,12 @@ def transport_from_sandbox(sandbox: SandboxHandle) -> ExecutionTransport:
         return ExecutionTransport(
             kind="local_namespace", workspace=str(workspace), pid=parsed_pid,
             owner_execute=sandbox._owner_execute,
+            owner_workload_execute=sandbox._owner_workload_execute,
             timeout_seconds=sandbox._timeout_seconds,
         )
     if has_remote:
+        if sandbox._credential_route_refs:
+            raise TransportError("SSH transport has no owner credential delivery contract")
         if not ssh_target or not remote_dir:
             raise TransportError(
                 "incomplete remote sandbox reachability: ssh and remote_dir are required"

tests/test_gateway.py

@@ -150,6 +150,7 @@ def test_owner_commands_preserve_create_identity_routes_and_timeout(request_id):
             assert "_owner_execute" not in sandbox.model_dump()
             transport = transport_from_sandbox(sandbox)
             transport.run(["printf", "%s", "non-secret"], timeout=30)
+            transport.run_workload(["rein-aharness", "run"], timeout=30)
             path = transport.write_task_file({"title": "private prompt"})
             transport.remove_file(path)
             return super().start_session(profile, inputs, sandbox)
@@ -164,14 +165,14 @@ def test_owner_commands_preserve_create_identity_routes_and_timeout(request_id):
     assert create_consumer.project == "exact-project"
     assert create_consumer.run_id == result.evidence.request_id
     assert create_consumer.session_id is None
-    assert manager.execute.call_count == 3
+    assert manager.execute.call_count == 4
     for call in manager.execute.call_args_list:
         sandbox_id, command = call.args
         assert sandbox_id == "sbx1"
         assert command.consumer == create_consumer
-        assert command.credential_route_refs == refs
+        assert command.credential_route_refs == (refs if command.command[0] == "rein-aharness" else [])
         assert command.timeout_seconds == 7
-    writer = manager.execute.call_args_list[1].args[1]
+    writer = manager.execute.call_args_list[2].args[1]
     assert writer.stdin_text == '{"title": "private prompt"}'
     assert "private prompt" not in str(writer.command)
     manager.destroy.assert_called_once_with("sbx1")

tests/test_rein_aharness.py

@@ -80,18 +80,18 @@ def test_dispatch_tool_invokes_agent_harness_cli() -> None:
         budget_tokens=1234,
     )
     transport = MagicMock(spec=ExecutionTransport)
-    transport.run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
+    transport.run_workload.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
     session = {"task_file": "/tmp/task.json", "transport": transport, "timeout_seconds": 30}
 
     with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
         result = rein.dispatch_tool(session, ToolCall(name="run_task"))
 
-    argv = transport.run.call_args.args[0]
+    argv = transport.run_workload.call_args.args[0]
     assert argv[:3] == ["/usr/bin/agent-harness", "run", "--task-file"]
     assert argv[argv.index("--model") + 1] == "claude-sonnet-4-6"
     assert argv[argv.index("--tool-profile") + 1] == "green-commit-only"
     assert argv[argv.index("--budget-tokens") + 1] == "1234"
-    assert transport.run.call_args.kwargs["timeout"] == 30
+    assert transport.run_workload.call_args.kwargs["timeout"] == 30
     assert result.ok is True
     assert result.output == "ok"
 
@@ -99,7 +99,7 @@ def test_dispatch_tool_invokes_agent_harness_cli() -> None:
 def test_dispatch_tool_reports_failure() -> None:
     rein = ReinAharness()
     transport = MagicMock(spec=ExecutionTransport)
-    transport.run.return_value = MagicMock(returncode=1, stdout="", stderr="boom")
+    transport.run_workload.return_value = MagicMock(returncode=1, stdout="", stderr="boom")
     session = {"task_file": "/tmp/task.json", "transport": transport, "timeout_seconds": 30}
 
     with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
@@ -167,14 +167,14 @@ def test_dispatch_tool_streams_events_when_enabled() -> None:
         json.dumps({"stream_event": {"type": "system", "subtype": "hook_started"}}),
         json.dumps({"ok": True, "committed": True}, indent=2),
     ]
-    transport.run.return_value = MagicMock(
+    transport.run_workload.return_value = MagicMock(
         returncode=0, stdout="\n".join(stdout_lines), stderr=""
     )
 
     with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
         result = rein.dispatch_tool(session, ToolCall(name="run_task"))
 
-    argv = transport.run.call_args.args[0]
+    argv = transport.run_workload.call_args.args[0]
     assert "--stream-tool-events" in argv
     assert len(result.events) == 2
     assert result.events[0]["message"]["content"][0]["name"] == "Read"
@@ -186,13 +186,13 @@ def test_dispatch_tool_streams_events_when_enabled() -> None:
 def test_dispatch_tool_no_stream_flag_by_default() -> None:
     rein = ReinAharness()
     transport = MagicMock(spec=ExecutionTransport)
-    transport.run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
+    transport.run_workload.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
     session = {"task_file": "/tmp/task.json", "transport": transport, "timeout_seconds": 30}
 
     with patch.object(rein, "_bin", return_value="/usr/bin/agent-harness"):
         result = rein.dispatch_tool(session, ToolCall(name="run_task"))
 
-    argv = transport.run.call_args.args[0]
+    argv = transport.run_workload.call_args.args[0]
     assert "--stream-tool-events" not in argv
     assert result.events == []
 
@@ -235,7 +235,7 @@ def test_resolved_model_requires_rein_report(reported) -> None:
     payload = {"ok": True}
     if reported is not None:
         payload["model"] = reported
-    transport.run.return_value = MagicMock(returncode=0, stdout=json.dumps(payload), stderr="")
+    transport.run_workload.return_value = MagicMock(returncode=0, stdout=json.dumps(payload), stderr="")
     transport.git_head.return_value = "new-head"
     session = {"task_file": "/tmp/task.json", "transport": transport,
                "timeout_seconds": 30, "head_before": "old-head"}

tests/test_rein_openweights.py

@@ -76,13 +76,13 @@ def test_start_session_requires_resolvable_target_repo() -> None:
 def test_dispatch_tool_invokes_rein_openweights_cli() -> None:
     rein = ReinOpenWeights()
     transport = MagicMock(spec=ExecutionTransport)
-    transport.run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
+    transport.run_workload.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
     session = {"task_file": "/tmp/task.json", "transport": transport, "timeout_seconds": 30}
 
     with patch.object(rein, "_bin", return_value="/usr/bin/rein-openweights"):
         result = rein.dispatch_tool(session, ToolCall(name="run_task"))
 
-    argv = transport.run.call_args.args[0]
+    argv = transport.run_workload.call_args.args[0]
     assert argv[:3] == ["/usr/bin/rein-openweights", "run", "--task-file"]
     assert "--no-hub" in argv
     assert result.ok is True
@@ -97,13 +97,13 @@ def test_dispatch_tool_passes_model_when_set() -> None:
         tool_profile="green-commit-only",
     )
     transport = MagicMock(spec=ExecutionTransport)
-    transport.run.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
+    transport.run_workload.return_value = MagicMock(returncode=0, stdout="ok", stderr="")
     session = {"task_file": "/tmp/task.json", "transport": transport, "timeout_seconds": 30}
 
     with patch.object(rein, "_bin", return_value="/usr/bin/rein-openweights"):
         rein.dispatch_tool(session, ToolCall(name="run_task"))
 
-    argv = transport.run.call_args.args[0]
+    argv = transport.run_workload.call_args.args[0]
     assert "--model" in argv
     assert "meta-llama/llama-3.1-70b-instruct" in argv
     assert argv[argv.index("--max-turns") + 1] == "8"
@@ -114,7 +114,7 @@ def test_dispatch_tool_passes_model_when_set() -> None:
 def test_dispatch_tool_reports_failure() -> None:
     rein = ReinOpenWeights()
     transport = MagicMock(spec=ExecutionTransport)
-    transport.run.return_value = MagicMock(returncode=1, stdout="", stderr="boom")
+    transport.run_workload.return_value = MagicMock(returncode=1, stdout="", stderr="boom")
     session = {"task_file": "/tmp/task.json", "transport": transport, "timeout_seconds": 30}
 
     with patch.object(rein, "_bin", return_value="/usr/bin/rein-openweights"):
@@ -127,7 +127,7 @@ def test_dispatch_tool_reports_failure() -> None:
 def test_dispatch_tool_enforces_outer_timeout() -> None:
     rein = ReinOpenWeights()
     transport = MagicMock(spec=ExecutionTransport)
-    transport.run.side_effect = subprocess.TimeoutExpired("rein-openweights", 30)
+    transport.run_workload.side_effect = subprocess.TimeoutExpired("rein-openweights", 30)
     session = {
         "task_file": "/tmp/task.json",
         "transport": transport,
@@ -138,7 +138,7 @@ def test_dispatch_tool_enforces_outer_timeout() -> None:
         with pytest.raises(subprocess.TimeoutExpired):
             rein.dispatch_tool(session, ToolCall(name="run_task"))
 
-    assert transport.run.call_args.kwargs["timeout"] == 30
+    assert transport.run_workload.call_args.kwargs["timeout"] == 30
 
 
 def test_end_session_detects_new_commit(tmp_path) -> None:
@@ -183,7 +183,7 @@ def test_resolved_model_requires_rein_report(reported) -> None:
     payload = {"ok": True}
     if reported is not None:
         payload["model"] = reported
-    transport.run.return_value = MagicMock(returncode=0, stdout=json.dumps(payload), stderr="")
+    transport.run_workload.return_value = MagicMock(returncode=0, stdout=json.dumps(payload), stderr="")
     transport.git_head.return_value = "new-head"
     session = {"task_file": "/tmp/task.json", "transport": transport,
                "timeout_seconds": 30, "head_before": "old-head"}

tests/test_rein_session_isolation.py

@@ -19,7 +19,7 @@ def test_interleaved_sessions_keep_their_own_outcomes_and_usage(adapter):
         transport = MagicMock(spec=ExecutionTransport)
         transport.resolve_executable.return_value = "/runtime/rein"
         transport.git_head.return_value = f"head-{index}"
-        transport.run.return_value = MagicMock(returncode=0 if ok else 1, stderr="",
+        transport.run_workload.return_value = MagicMock(returncode=0 if ok else 1, stderr="",
             stdout=json.dumps({"ok": ok, "tokens_spent": 10 + index,
                                "model": f"model-{index}", "reason": None if ok else "failed"}))
         sessions.append({"transport": transport, "task_file": f"/workspace-{index}/task",

tests/test_transport.py

@@ -173,3 +173,16 @@ def test_git_head_fails_closed_on_transport_error() -> None:
     ):
         with pytest.raises(TransportError, match="inside the selected sandbox"):
             transport.git_head()
+
+
+def test_workload_refuses_missing_dedicated_owner_binding():
+    transport = ExecutionTransport(kind="local_namespace", workspace="/work", owner_execute=lambda *args: None)
+    with pytest.raises(TransportError, match="workload binding"):
+        transport.run_workload(["true"])
+
+
+def test_remote_transport_refuses_requested_credential_routes():
+    sandbox = SandboxHandle(sandbox_id="remote", host="remote", reachability={"ssh":"agent@host", "remote_dir":"/work"})
+    sandbox._credential_route_refs = ["provider-route"]
+    with pytest.raises(TransportError, match="credential delivery"):
+        transport_from_sandbox(sandbox)

workplans/GLAS-WP-0012-first-local-profile-production-proof.md

@@ -423,3 +423,28 @@ reload after real resume/compaction, provider/repository changes and exact
 available model/session metadata. The current lifecycle exposes no continuity
 event API. Existing owner-delivery blockers and all task/readiness states remain;
 local deterministic tests are not governed provider or isolation acceptance.
+
+
+## 2026-10-16 NetKingdom isolated integration preparation
+
+Prepared, without owner deployment or readiness promotion, a Glas change that
+separates credential-free housekeeping commands from explicit rein workload
+execution. Both concrete adapters use the workload path; SSH selections with
+credential routes refuse because that transport has no owner delivery contract.
+Full regression: 141 passed. Real disposable bwrap/Glas transport proof passes
+synthetic host-file denial, inherited credential environment and subprocess
+denial, workload-only delivery, redaction, subsequent-command absence and
+teardown. No provider/model call or real credential read occurred.
+
+Latest owner dependency sources supersede the stale missing-lane account:
+September 27 native lanes were activated and revoked; corrected immutable
+owner runtime is `818b1b7c…` (October 15 startup proof), not yet installed or
+admitted on Railiance. REINAH-WP-0008-T07 retains memo lifecycle acceptance,
+installation, rehearsal, renewal/pin updates, native approvals and bounded
+successful model/tool/commit acceptance. Any subsequent build containing this
+patch has a different source/artifact pin requiring owner admission.
+
+This is a reviewable patch in Custodian's NetKingdom change bundle; no canonical
+source synchronization or workplan task completion is implied. CUST-WP-0073
+continues to require admission of the actual interactive assignment, including
+its existing host credential and external tool paths, not a disposable fixture.