Generalise the harness and close the defects that made runs look verified

.gitignore

@@ -21,3 +21,6 @@ examples/artifacts/
 
 # Daytona snapshot SDK venv (scripts/daytona-snapshot.sh)
 .venv-daytona/
+
+# Local test-suite output, regenerated on every run.
+recordings/

Dockerfile.hosted

@@ -72,7 +72,7 @@ WORKDIR /opt/alk
 # Daytona's direct-image builder can reuse an image when only build-context
 # files change.  Bump this source revision whenever guest code changes so a
 # hosted test cannot silently execute an older installed ALK.
-ARG ALK_HOSTED_SOURCE_REVISION=20260829-visible-diagnostics-r12
+ARG ALK_HOSTED_SOURCE_REVISION=20260830-r1
 LABEL io.futureagi.alk-source-revision="${ALK_HOSTED_SOURCE_REVISION}"
 COPY pyproject.toml README.md ./
 COPY src ./src

src/fi/alk/harness/HOW-IT-WORKS.md

@@ -99,21 +99,26 @@ migrations, seed process and tool code run in an isolated environment. A call fo
 is not there is refused by the submitted implementation, not by a mock or a rewritten handler.
 
 It builds three things, all shared by every scenario: **the world**, **the simulator prompt** for
-a conversational agent, and **the sub-goal catalogue**. The stage has sixteen tools and no file
-access at all:
+a conversational agent, and **the sub-goal catalogue**. The stage has twenty-one tools, and a
+shell and an editor besides: building infrastructure and proving it answers is engineering, and
+withholding those did not make it safer, only unable to finish. What it may use is still only
+what it was granted, which is enforced on every call rather than assumed:
 
 | Tool | Does |
 |---|---|
 | `create_schema` | Run the CREATE TABLE statements |
 | `seed` | Insert rows — the agent's real catalogue |
 | `change_data` | One UPDATE or DELETE, for fixing a row put in wrong |
 | `adopt_tool` | Bind and smoke-test the agent's own implementation |
-| `write_env_file` / `run_env_command` | Container orchestration only; never agent behavior |
+| `adopt_store` / `adopt_state` | Bind the store or in-process state the repository already has |
+| `write_store_ops` | Write the read and write operations the world is driven through |
 | `run_tool` | Call a defined tool and see what the world does |
 | `declare_sequence` / `drop_sequence` | A series of calls whose end state must hold |
 | `inspect_world` | Look at what is in the world |
 | `amend_contract`, `add_rule`, `drop_rule`, `fix_tool` | Correct the contract |
 | `check_world` | Run every probe, report without saving |
+| `set_modality` / `write_simulator_prompt` | How the agent is reached, and the caller it faces |
+| `add_sub_goal` / `add_world_check` | What a run has to achieve, and what must hold after it |
 | `save_world` | Freeze it — refused unless it holds up |
 
 Bindings are small adapters to the callable the repository ships. There is no generated-handler

src/fi/alk/harness/authoring_entrypoint.py

@@ -11,12 +11,90 @@
 import argparse
 import asyncio
 import json
+import logging
 from pathlib import Path
 
+from .build import refusal_at
 from .cli import _auto
-from .job import HarnessJob
+from .job import FailureDomain, HarnessJob, HarnessStage
 from .scenarios import load as load_written
 
+logger = logging.getLogger(__name__)
+
+REFUSAL_CODE = "environment_not_buildable"
+
+
+async def _report_refusal(problems: list[str]) -> bool:
+    """Tell the platform the environment stage declined, and why, before this process exits.
+
+    Nothing downstream will do it. The guest runs `authoring && bundle && run` as one shell
+    chain, so a non-zero authoring exit short-circuits it and the run entrypoint -- the only
+    component that owns an outbound channel -- never starts. What the control plane is left with
+    is an exit code, which it reports as `guest_crashed` in the `infrastructure` domain: a
+    principled refusal presented as a crashed sandbox, sending an operator to look at Daytona,
+    the image and the network, all of which are healthy, while the actual remedy sits in a log
+    they have no reason to open. `infrastructure` is also a retryable domain, so the same correct
+    refusal gets re-derived in a second sandbox.
+
+    The domain here is `agent`: the submitted repository ships no seam to build against. That is
+    not in the platform's retryable set, so this cannot be retried into the same answer twice.
+    """
+    from . import outbound as ob
+    from .hosted_entrypoint import HostedEntrypointDeps, OutboundAdapter
+
+    deps = HostedEntrypointDeps()
+    try:
+        capabilities = deps.load_capabilities()
+    except ob.CapabilitiesError as exc:
+        # No channel: the ordinary shape of a local run, and not an error. Hosted runs always
+        # have one, so this staying quiet locally does not hide anything hosted.
+        logger.info("no outbound channel to report the refusal through: %s", exc.code)
+        return False
+
+    work_directory = Path("/work")
+    channel_state = ob.ChannelState()
+    transport = deps.build_transport()
+    retry_policy = deps.retry_policy()
+    events_spool = deps.build_events_spool(work_directory)
+    adapter = OutboundAdapter(
+        capabilities,
+        events_spool=events_spool,
+        events_client=ob.EventsClient(
+            capabilities,
+            events_spool,
+            transport,
+            retry_policy=retry_policy,
+            channel_state=channel_state,
+        ),
+        results_client=ob.ResultsClient(
+            capabilities, transport, retry_policy=retry_policy, channel_state=channel_state
+        ),
+        artifacts_client=ob.ArtifactsClient(
+            capabilities, transport, retry_policy=retry_policy, channel_state=channel_state
+        ),
+        channel_state=channel_state,
+        extra_secret_values=deps.peek_secret_values(),
+    )
+    # The remedy travels in `message`, because the terminal event's failure shape is
+    # {domain, stage, code, message} with extra="forbid" and has nowhere else to put it. Every
+    # line names one tool and what the repository must expose for it, which is the whole value.
+    await adapter.emit_terminal(
+        stage=HarnessStage.FAILED,
+        failure={
+            "domain": FailureDomain.AGENT.value,
+            "stage": HarnessStage.GENERATING_ENVIRONMENT.value,
+            "code": REFUSAL_CODE,
+            "message": (
+                "The environment stage declined to build: the submitted repository does not "
+                "expose a runnable seam for these tools, and building one would mean inventing "
+                "agent behaviour, which would grade nothing and look green.\n  - "
+                + "\n  - ".join(problems)
+            ),
+        },
+    )
+    await adapter.drain(complete=False, deadline=adapter.deadline())
+    return True
+
 
 def _persist_authored_scenario_count(
     job_path: Path, job: HarnessJob, output: Path
@@ -74,6 +152,13 @@ def main(argv: list[str] | None = None) -> int:
     status = asyncio.run(_auto(namespace))
     if status == 0:
         _persist_authored_scenario_count(args.job, job, args.output.resolve())
+        return status
+    problems = refusal_at(args.output.resolve())
+    if problems:
+        try:
+            asyncio.run(_report_refusal(problems))
+        except Exception:  # noqa: BLE001 - reporting must never replace the refusal itself
+            logger.exception("could not report the environment refusal upward")
     return status
 
 

src/fi/alk/harness/backends/base.py

@@ -102,9 +102,9 @@ class SessionSpec:
     ``builtins`` are host tools by bare name (``Read``, ``Glob``, ``Grep``,
     ``AskUserQuestion``); ``servers`` are the harness's own tools. ``ask`` is the operator
     callback consulted when the model asks a question; None means the run is unattended.
-    ``gated`` selects the deny-by-default permission regime every tool-bearing stage runs
-    under; the one stage that runs bare (the simulated customer, which has no tools) turns it
-    off to keep its behaviour byte-identical.
+    ``gated`` selects the deny-by-default regime a tool-bearing stage runs under: the tools it
+    was granted, and nothing the host happens to also expose. The one stage that runs bare (the
+    simulated customer, which has no tools) turns it off to keep its behaviour byte-identical.
     ``thinking`` opts into the harness's thinking policy (config.thinking_config); stages that
     never set one keep their backend's default.
     """

src/fi/alk/harness/backends/claude.py

@@ -183,7 +183,14 @@ def create(self, spec: SessionSpec) -> ClaudeSession:
             # is consulted, so a stage could rewrite an artifact by hand and skip the tool whose
             # whole job is to validate that change.
             options.permission_mode = "default"
-            options.disallowed_tools = list(UNWANTED)
+            # Deny only what this stage was not granted. A stage that asks for Bash or Write
+            # means it, and a blanket denial here would silently outrank its own tool list.
+            options.disallowed_tools = [
+                name for name in UNWANTED if name not in set(allowed)
+            ]
+            # The hook is the enforcement; the callback is the backstop and the question route.
+            # allowed_tools shadows the callback for everything granted, so the hook is the only
+            # thing consulted on every call.
             options.hooks = gate_hooks(allowed)
             options.can_use_tool = spec.permission_override or permission_gate(
                 spec.ask, allowed

src/fi/alk/harness/build.py

@@ -11,6 +11,8 @@
 from __future__ import annotations
 
 import asyncio
+import json
+import logging
 import os
 from collections.abc import Callable
 from pathlib import Path
@@ -23,6 +25,8 @@
 from .world.snapshot import saved as world_saved
 from .world.tools import WORLD_SERVER, world_tools
 
+logger = logging.getLogger(__name__)
+
 SKILL = "build-environment"
 
 
@@ -80,15 +84,62 @@ def blockers(contract: AgentContract, source_root: str = "") -> list[str]:
     return problems
 
 
-def require_buildable(contract: AgentContract, source_root: str = "") -> None:
-    problems = blockers(contract, source_root)
-    if problems:
-        raise RuntimeError(
+REFUSAL_DOCUMENT = "environment-refusal.json"
+
+
+class EnvironmentNotBuildable(RuntimeError):
+    """The repository does not ship a seam the environment could be built against.
+
+    Its own type, and carrying its problems as data rather than only as a formatted message,
+    because this is the refusal an operator can actually act on: every entry names one tool and
+    what the repository must expose for it. Flattened into a string it becomes a log line
+    somebody has to go looking for, and the process that has to report it upward sees only an
+    exit code, which is indistinguishable from the harness falling over.
+    """
+
+    def __init__(self, problems: list[str]) -> None:
+        self.problems = list(problems)
+        super().__init__(
             "Cannot create a truthful test environment without reimplementing agent behavior:\n"
             "  - " + "\n  - ".join(problems)
         )
 
 
+def record_refusal(destination: Path, problems: list[str]) -> Path:
+    """Leave the refusal where the process that must report it can find it.
+
+    The stage that decides this and the process that reports it upward are different processes,
+    and what crosses that boundary is an exit status. A non-zero exit is read by everything above
+    as "the guest crashed", so the reason has to travel as a document or it does not travel.
+    """
+    path = destination / REFUSAL_DOCUMENT
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps({"problems": list(problems)}, indent=2) + "\n", encoding="utf-8"
+    )
+    return path
+
+
+def refusal_at(destination: Path) -> list[str]:
+    """The recorded refusal, or nothing. Unreadable is not the same as absent, and says so."""
+    path = destination / REFUSAL_DOCUMENT
+    if not path.is_file():
+        return []
+    try:
+        body = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as broke:
+        logger.warning("%s exists but could not be read: %s", path, broke)
+        return []
+    problems = body.get("problems") if isinstance(body, dict) else None
+    return [str(one) for one in problems] if isinstance(problems, list) else []
+
+
+def require_buildable(contract: AgentContract, source_root: str = "") -> None:
+    problems = blockers(contract, source_root)
+    if problems:
+        raise EnvironmentNotBuildable(problems)
+
+
 def turns_for(contract: AgentContract) -> int:
     """A turn budget that grows with the agent being built for.
 
@@ -195,7 +246,7 @@ def open_stage(
             "environment endpoints: "
             + (", ".join(runtime_tools) or "none")
             + "\nDo not adopt either group, inspect their source again, or recreate any service "
-            "or behavior. Do not use run_env_command for source discovery. The contract already "
+            "or behavior. Do not go source hunting with the shell. The contract already "
             "contains that evidence. Inspect the live data once. Preserve useful repository seed "
             "rows. If the submitted schema is empty or lacks the records needed to exercise the "
             "contract's branches, add a small varied realistic baseline through seed only; never "
@@ -237,10 +288,21 @@ def open_stage(
             f"{load_skill(SKILL)}\n\n## This agent\n\n{contract.brief(with_data=True)}"
             + environment_note
         ),
-        # No file tools and no shell. Everything this stage can do goes through a tool that
-        # executes it and reports back, which is what makes the guardrails meaningful.
         servers={WORLD_SERVER: server},
-        builtins=("AskUserQuestion",),
+        # A shell, a file editor and the repository. This stage builds infrastructure for an
+        # agent it has never seen, so the work is engineering rather than form filling: read the
+        # code, write what it needs, run it, read the error, fix it. This grant is the boundary,
+        # and it is the same one hosted and locally: a sandbox contains the hosted lane, but the
+        # same stage runs in-process on an operator's machine, where nothing contains a shell.
+        builtins=(
+            "AskUserQuestion",
+            "Read",
+            "Glob",
+            "Grep",
+            "Write",
+            "Edit",
+            "Bash",
+        ),
         cwd=str(destination.parent if destination.parent.exists() else Path.cwd()),
         max_turns=environment_turns_for(
             contract,