feat: add versioned execution profiles

src/glas_harness/channels/base.py

@@ -17,7 +17,7 @@ from typing import Any
 class GatewayInvocation:
     """Everything run_task_through_rein needs, channel-agnostic."""
 
-    sandbox_profile: str
+    harness_profile: str
     repo: str
     title: str
     description: str

src/glas_harness/channels/cli_channel.py

@@ -18,7 +18,7 @@ from glas_harness.channels.base import Channel, GatewayInvocation
 class CLIChannel(Channel):
     def parse_invocation(self, raw: argparse.Namespace) -> GatewayInvocation:
         return GatewayInvocation(
-            sandbox_profile=raw.sandbox_profile,
+            harness_profile=raw.harness_profile,
             repo=raw.repo,
             title=raw.title,
             description=raw.description,

src/glas_harness/cli.py

@@ -10,8 +10,14 @@ def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(prog="glas-harness")
     sub = parser.add_subparsers(dest="command", required=True)
 
-    run = sub.add_parser("run", help="Run one task through a rein inside a sand-boxer sandbox")
-    run.add_argument("--sandbox-profile", required=True, help="e.g. profile.bwrap-local")
+    run = sub.add_parser(
+        "run", help="Run one task through a versioned Glas execution profile"
+    )
+    run.add_argument(
+        "--harness-profile",
+        required=True,
+        help="Explicit profile id[@version], e.g. harness.agent-dev-local@1.0.0",
+    )
     run.add_argument("--repo", required=True, help="Local repo path to mirror into the sandbox")
     run.add_argument("--title", required=True)
     run.add_argument("--description", required=True)
@@ -19,25 +25,56 @@ def main(argv: list[str] | None = None) -> int:
     run.add_argument("--project", default="glas-harness")
     run.add_argument("--no-hub", action="store_true", help="Skip the gateway's own hub reporting")
 
+    profiles = sub.add_parser("profiles", help="Validate and list executable Glas profiles")
+    profiles.add_argument("--json", action="store_true", help="Emit machine-readable JSON")
+
     args = parser.parse_args(argv)
 
     if args.command == "run":
         from glas_harness.channels.cli_channel import CLIChannel
-        from glas_harness.gateway import run_task_through_rein
+        from glas_harness.contract import ExecutionRequest
+        from glas_harness.gateway import run_execution
 
         channel = CLIChannel()
         invocation = channel.parse_invocation(args)
-        result = run_task_through_rein(
-            sandbox_profile=invocation.sandbox_profile,
-            repo=invocation.repo,
-            title=invocation.title,
-            description=invocation.description,
-            actor=invocation.actor,
-            project=invocation.project,
-            report_to_hub=invocation.report_to_hub,
+        result = run_execution(
+            ExecutionRequest(
+                harness_profile_ref=invocation.harness_profile,
+                repo=invocation.repo,
+                title=invocation.title,
+                description=invocation.description,
+                actor=invocation.actor,
+                project=invocation.project,
+                report_to_hub=invocation.report_to_hub,
+            )
         )
-        print(channel.render_result(result))
-        return 0 if result["tool_ok"] else 1
+        print(channel.render_result(result.model_dump(mode="json")))
+        return 0 if result.ok else 1
+
+    if args.command == "profiles":
+        import json
+
+        from glas_harness.profiles import ProfileCatalog, ProfileError
+
+        try:
+            rows = [
+                context.model_dump(mode="json")
+                for context in ProfileCatalog().validate_all()
+            ]
+        except ProfileError as exc:
+            print(f"invalid profile catalog: {exc}", file=sys.stderr)
+            return 2
+        if args.json:
+            print(json.dumps(rows, indent=2))
+        else:
+            for row in rows:
+                print(
+                    f"{row['profile']['id']}@{row['profile']['version']}\t"
+                    f"rein={row['rein_id']}@{row['rein_version']}\t"
+                    f"model={row['model']['model']}\t"
+                    f"sandbox={row['sandbox_profile']}"
+                )
+        return 0
 
     return 1
 

src/glas_harness/contract.py

@@ -1,47 +1,202 @@
-"""The harness contract a concrete rein implements.
+"""Versioned contract between glas-harness and concrete reins.
 
-See docs/harness-contract.md for the full session-lifecycle diagram and
-GLAS-WP-0001-T01. glas-harness owns the outer loop (profile resolution,
-sandbox request/teardown via sand-boxer, State Hub reporting, actor
-attribution); a rein owns the inner agentic loop.
+Glas owns the outer lifecycle and the stable boundary models. A rein owns its
+inner agentic loop. Contract models reject unknown fields so a profile or
+backend cannot silently widen the execution surface.
 """
 
 from __future__ import annotations
 
+import re
 from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from typing import Any
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+CONTRACT_VERSION = "1.0"
+_SEMVER = re.compile(r"^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:[-+][0-9A-Za-z.-]+)?$")
 
 
-@dataclass
-class SandboxHandle:
-    """What glas-harness got back from sand-boxer's create()."""
+class ContractModel(BaseModel):
+    model_config = ConfigDict(extra="forbid")
 
+
+class HarnessProfileRef(ContractModel):
+    id: str = Field(pattern=r"^harness\.[a-z0-9][a-z0-9._-]*$")
+    version: str
+
+    @field_validator("version")
+    @classmethod
+    def validate_version(cls, value: str) -> str:
+        if not _SEMVER.match(value):
+            raise ValueError("profile version must be semantic versioning")
+        return value
+
+    def __str__(self) -> str:
+        return f"{self.id}@{self.version}"
+
+
+class ModelRoute(ContractModel):
+    provider: str = Field(min_length=1)
+    model: str = Field(min_length=1)
+    model_class: Literal["frontier", "open-weight", "specialized", "other"]
+    route: str = Field(default="rein-native", min_length=1)
+
+
+class ExecutionLimits(ContractModel):
+    budget_tokens: int | None = Field(default=None, gt=0)
+    timeout_seconds: int | None = Field(default=None, gt=0)
+    max_turns: int | None = Field(default=None, gt=0)
+
+
+class ReinSelection(ContractModel):
+    id: str = Field(pattern=r"^rein-[a-z0-9][a-z0-9-]*$")
+    required_capabilities: dict[str, Any] = Field(default_factory=dict)
+
+
+class HarnessProfile(ContractModel):
+    id: str = Field(pattern=r"^harness\.[a-z0-9][a-z0-9._-]*$")
+    version: str
+    contract_version: str
+    status: Literal["enabled", "disabled"] = "enabled"
+    rein: ReinSelection
+    sandbox_profile: str = Field(pattern=r"^profile\.[a-z0-9][a-z0-9._-]*$")
+    tool_profile: str = Field(min_length=1)
+    model: ModelRoute
+    limits: ExecutionLimits = Field(default_factory=ExecutionLimits)
+    credential_route_refs: list[str] = Field(default_factory=list)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("version")
+    @classmethod
+    def validate_version(cls, value: str) -> str:
+        return HarnessProfileRef.validate_version(value)
+
+    @property
+    def ref(self) -> HarnessProfileRef:
+        return HarnessProfileRef(id=self.id, version=self.version)
+
+
+class ReinDescriptor(ContractModel):
+    id: str = Field(pattern=r"^rein-[a-z0-9][a-z0-9-]*$")
+    version: str
+    title: str
+    description: str = ""
+    handler: str = Field(pattern=r"^glas_harness\.reins\.[A-Za-z0-9_]+:[A-Za-z0-9_]+$")
+    contract_versions: list[str] = Field(min_length=1)
+    capabilities: dict[str, Any] = Field(default_factory=dict)
+    status: Literal["implemented", "disabled", "experimental"] = "implemented"
+
+    @field_validator("version")
+    @classmethod
+    def validate_version(cls, value: str) -> str:
+        return HarnessProfileRef.validate_version(value)
+
+
+class SandboxHandle(ContractModel):
     sandbox_id: str
     host: str
-    reachability: dict[str, Any] = field(default_factory=dict)
+    reachability: dict[str, Any] = Field(default_factory=dict)
 
 
-@dataclass
-class ToolCall:
+class ToolCall(ContractModel):
     name: str
-    args: dict[str, Any] = field(default_factory=dict)
+    args: dict[str, Any] = Field(default_factory=dict)
     actor: str = "agt"
 
 
-@dataclass
-class ToolResult:
+class ToolResult(ContractModel):
     ok: bool
     output: str = ""
     error: str | None = None
-    # Per-tool-call audit trail, populated when the rein can observe its own
-    # inner loop's individual tool invocations in real time (e.g. parsing
-    # `claude --output-format stream-json`). Empty when a rein only exposes
-    # its whole task as one opaque call — that's a real limit of CLI-wrapped
-    # agents, not a shortcut: neither Claude Code nor rein-openweights's own
-    # loop lets glas-harness externally execute individual tool calls, only
-    # observe them. See HARNESS-WP-0002-T03.
-    events: list[dict[str, Any]] = field(default_factory=list)
+    events: list[dict[str, Any]] = Field(default_factory=list)
+    events_completeness: Literal["complete", "partial", "unavailable"] = "unavailable"
+    tokens_spent: int | None = Field(default=None, ge=0)
+    duration_s: float | None = Field(default=None, ge=0)
+    resolved_model: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class ExecutionSummary(ContractModel):
+    committed: bool
+    commit_sha: str | None = None
+    outcome: Literal["succeeded", "failed", "refused"]
+    reason: str | None = None
+    tokens_spent: int | None = Field(default=None, ge=0)
+    duration_s: float | None = Field(default=None, ge=0)
+    resolved_model: str | None = None
+    artifacts: list[str] = Field(default_factory=list)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class ExecutionRequest(ContractModel):
+    harness_profile_ref: str = Field(min_length=1)
+    repo: str = Field(min_length=1)
+    title: str = Field(min_length=1)
+    description: str
+    actor: str = "agt"
+    project: str = "glas-harness"
+    request_id: str | None = None
+    correlation_id: str | None = None
+    assignment_ref: str | None = None
+    role_ref: str | None = None
+    duty_ref: str | None = None
+    goal_refs: list[str] = Field(default_factory=list)
+    resource_envelope_refs: list[str] = Field(default_factory=list)
+    expected_output: str | None = None
+    report_to_hub: bool = True
+
+
+class ResolvedExecutionContext(ContractModel):
+    contract_version: str
+    profile: HarnessProfileRef
+    rein_id: str
+    rein_version: str
+    sandbox_profile: str
+    tool_profile: str
+    model: ModelRoute
+    limits: ExecutionLimits
+
+
+class ExecutionEvidence(ContractModel):
+    request_id: str
+    correlation_id: str | None = None
+    actor: str
+    project: str
+    target_repo: str
+    contract_version: str = CONTRACT_VERSION
+    profile_ref: str | None = None
+    rein_id: str | None = None
+    rein_version: str | None = None
+    model_route: str | None = None
+    resolved_model: str | None = None
+    sandbox_profile: str | None = None
+    sandbox_id: str | None = None
+    tool_profile: str | None = None
+    outcome: Literal["succeeded", "failed", "refused"]
+    failure_stage: Literal[
+        "resolution", "sandbox_create", "session_start", "execution", "session_end", "teardown"
+    ] | None = None
+    error: str | None = None
+    started_at: str
+    finished_at: str
+    duration_s: float = Field(ge=0)
+    tokens_spent: int | None = Field(default=None, ge=0)
+    token_budget: int | None = Field(default=None, ge=0)
+    commit_sha: str | None = None
+    artifacts: list[str] = Field(default_factory=list)
+    tool_events_count: int = Field(default=0, ge=0)
+    tool_events_completeness: Literal["complete", "partial", "unavailable"] = "unavailable"
+    refs: dict[str, Any] = Field(default_factory=dict)
+
+
+class GatewayResult(ContractModel):
+    ok: bool
+    evidence: ExecutionEvidence
+    # Direct caller output only. gateway.py deliberately excludes these fields
+    # from State Hub detail because they can contain prompts/model responses.
+    tool_output: str = ""
+    tool_error: str | None = None
 
 
 class Rein(ABC):
@@ -49,14 +204,14 @@ class Rein(ABC):
 
     @abstractmethod
     def start_session(
-        self, profile: dict[str, Any], inputs: dict[str, str], sandbox: SandboxHandle
+        self, profile: HarnessProfile, inputs: dict[str, str], sandbox: SandboxHandle
     ) -> dict[str, str]:
-        """Begin an agent session bound to a sandbox. Returns a session handle."""
+        """Begin an agent session bound to a sandbox."""
 
     @abstractmethod
     def dispatch_tool(self, session: dict[str, str], tool_call: ToolCall) -> ToolResult:
-        """Run one tool call under the session's policy."""
+        """Run one tool call under the session policy."""
 
     @abstractmethod
-    def end_session(self, session: dict[str, str]) -> dict[str, str]:
-        """Close the session. Returns a summary (commit sha, outcome, ...)."""
+    def end_session(self, session: dict[str, str]) -> ExecutionSummary:
+        """Close the session and return normalized rein evidence."""

src/glas_harness/gateway.py

@@ -1,37 +1,201 @@
-"""Minimal gateway proving the harness contract against rein-aharness.
+"""Profile-driven Glas gateway.
 
-GLAS-WP-0001-T04: resolve a sand-boxer profile, request a sandbox,
-dispatch one rein-aharness task through the Rein contract, verify a
-commit landed, tear the sandbox down. This is the parity proof gating
-any later "retire rein-aharness as a standalone concern" conversation —
-it is not itself that conversation.
-
-GLAS-WP-0002-T03: post the gateway's own State Hub event, independent of
-whatever the rein itself reports (both reins' CLIs are invoked with
-their own hub reporting disabled by their glas-harness adapters — see
-reins/rein_aharness.py / reins/rein_openweights.py). Reported on both
-success and failure, from a `finally` block, so a raised exception still
-leaves an audit trail.
-
-Requires the `sandbox` extra (sand-boxer installed as a sibling
-editable dependency).
+The gateway resolves a versioned harness profile before creating a sandbox.
+There is deliberately no default rein or model in governed execution.
 """
 
 from __future__ import annotations
 
-from typing import Any
+import time
+import uuid
+from datetime import UTC, datetime
 
 from sandboxer.core.manager import SandboxManager
 from sandboxer.models import Consumer, SandboxCreateRequest
 
 from glas_harness import hub
-from glas_harness.contract import Rein, SandboxHandle, ToolCall
-from glas_harness.reins.rein_aharness import ReinAharness
+from glas_harness.contract import (
+    CONTRACT_VERSION,
+    ExecutionEvidence,
+    ExecutionRequest,
+    ExecutionSummary,
+    GatewayResult,
+    Rein,
+    SandboxHandle,
+    ToolCall,
+    ToolResult,
+)
+from glas_harness.profiles import ProfileCatalog
+
+
+def _now() -> str:
+    return datetime.now(UTC).isoformat()
+
+
+def _request_refs(request: ExecutionRequest) -> dict:
+    refs = {
+        "assignment_ref": request.assignment_ref,
+        "role_ref": request.role_ref,
+        "duty_ref": request.duty_ref,
+        "goal_refs": request.goal_refs,
+        "resource_envelope_refs": request.resource_envelope_refs,
+        "expected_output": request.expected_output,
+    }
+    return {key: value for key, value in refs.items() if value not in (None, [], "")}
+
+
+def run_execution(
+    request: ExecutionRequest,
+    *,
+    catalog: ProfileCatalog | None = None,
+    rein: Rein | None = None,
+    manager: SandboxManager | None = None,
+) -> GatewayResult:
+    """Resolve and run one request, returning evidence for every outcome."""
+
+    request_id = request.request_id or str(uuid.uuid4())
+    started_at = _now()
+    started = time.monotonic()
+    catalog = catalog or ProfileCatalog()
+    refs = _request_refs(request)
+
+    profile = None
+    descriptor = None
+    sandbox_id: str | None = None
+    tool_result: ToolResult | None = None
+    summary: ExecutionSummary | None = None
+    outcome = "failed"
+    failure_stage = None
+    error: str | None = None
+
+    try:
+        profile, descriptor = catalog.resolve(request.harness_profile_ref)
+        selected_rein = rein or catalog.build_rein(profile, descriptor)
+    except Exception as exc:
+        outcome = "refused"
+        failure_stage = "resolution"
+        error = str(exc)
+        result = _build_result(
+            request=request,
+            request_id=request_id,
+            started_at=started_at,
+            started=started,
+            outcome=outcome,
+            failure_stage=failure_stage,
+            error=error,
+            profile=profile,
+            descriptor=descriptor,
+            sandbox_id=None,
+            tool_result=None,
+            summary=None,
+            refs=refs,
+        )
+        _report(request, result)
+        return result
+
+    manager = manager or SandboxManager()
+    status = None
+    try:
+        try:
+            status = manager.create(
+                SandboxCreateRequest(
+                    profile=profile.sandbox_profile,
+                    inputs={"repo": request.repo},
+                    consumer=Consumer(actor=request.actor, project=request.project),
+                    ttl=None,
+                )
+            )
+            sandbox_id = status.sandbox_id
+        except Exception as exc:
+            failure_stage = "sandbox_create"
+            error = str(exc)
+            raise
+
+        reachability = (
+            status.reachability.model_dump(mode="json", exclude_none=True)
+            if status.reachability
+            else {}
+        )
+        sandbox = SandboxHandle(
+            sandbox_id=status.sandbox_id,
+            host=status.host or "",
+            reachability=reachability,
+        )
+
+        try:
+            session = selected_rein.start_session(
+                profile,
+                {
+                    "title": request.title,
+                    "description": request.description,
+                    "target_repo": request.repo,
+                    "request_id": request_id,
+                },
+                sandbox,
+            )
+        except Exception as exc:
+            failure_stage = "session_start"
+            error = str(exc)
+            raise
+
+        try:
+            tool_result = selected_rein.dispatch_tool(
+                session, ToolCall(name="run_task", actor=request.actor)
+            )
+        except Exception as exc:
+            failure_stage = "execution"
+            error = str(exc)
+            raise
+
+        try:
+            summary = selected_rein.end_session(session)
+        except Exception as exc:
+            failure_stage = "session_end"
+            error = str(exc)
+            raise
+
+        if tool_result.ok and summary.outcome == "succeeded":
+            outcome = "succeeded"
+        else:
+            outcome = summary.outcome if summary.outcome in {"failed", "refused"} else "failed"
+            failure_stage = "execution"
+            error = tool_result.error or summary.reason or "rein reported unsuccessful execution"
+    except Exception:
+        # The exact error/stage is captured above. All paths still tear down and
+        # return normalized evidence rather than leaking a provider exception.
+        pass
+    finally:
+        if status is not None:
+            try:
+                manager.destroy(status.sandbox_id)
+            except Exception as exc:
+                if outcome == "succeeded" or not error:
+                    outcome = "failed"
+                    failure_stage = "teardown"
+                    error = str(exc)
+
+    result = _build_result(
+        request=request,
+        request_id=request_id,
+        started_at=started_at,
+        started=started,
+        outcome=outcome,
+        failure_stage=failure_stage,
+        error=error,
+        profile=profile,
+        descriptor=descriptor,
+        sandbox_id=sandbox_id,
+        tool_result=tool_result,
+        summary=summary,
+        refs=refs,
+    )
+    _report(request, result)
+    return result
 
 
 def run_task_through_rein(
     *,
-    sandbox_profile: str,
+    harness_profile: str,
     repo: str,
     title: str,
     description: str,
@@ -39,89 +203,113 @@ def run_task_through_rein(
     actor: str = "agt",
     project: str = "glas-harness",
     manager: SandboxManager | None = None,
+    catalog: ProfileCatalog | None = None,
     report_to_hub: bool = True,
-) -> dict[str, Any]:
-    """Resolve `sandbox_profile`, run one task inside it via `rein`, verify, tear down.
+) -> dict:
+    """Compatibility-shaped wrapper around the versioned execution request.
 
-    Defaults to `ReinAharness` when no rein is supplied — the only
-    implemented rein as of GLAS-WP-0001-T04. `rein-openweights` plugs in
-    the same way once REIN-OW-WP-0001 lands.
+    It intentionally requires a harness profile. Direct rein injection is
+    retained only as a library/test seam and never selects a default backend.
     """
-    manager = manager or SandboxManager()
-    rein = rein or ReinAharness()
 
-    request = SandboxCreateRequest(
-        profile=sandbox_profile,
-        inputs={"repo": repo},
-        consumer=Consumer(actor=actor, project=project),
+    result = run_execution(
+        ExecutionRequest(
+            harness_profile_ref=harness_profile,
+            repo=repo,
+            title=title,
+            description=description,
+            actor=actor,
+            project=project,
+            report_to_hub=report_to_hub,
+        ),
+        catalog=catalog,
+        rein=rein,
+        manager=manager,
     )
-    status = manager.create(request)
-    result: dict[str, Any] | None = None
-    error: str | None = None
-    try:
-        reachability = status.reachability.model_dump(mode="json") if status.reachability else {}
-        sandbox = SandboxHandle(
-            sandbox_id=status.sandbox_id, host=status.host or "", reachability=reachability
-        )
-
-        session = rein.start_session(
-            profile={"id": sandbox_profile},
-            inputs={"title": title, "description": description},
-            sandbox=sandbox,
-        )
-        tool_result = rein.dispatch_tool(session, ToolCall(name="run_task", actor=actor))
-        summary = rein.end_session(session)
-
-        result = {
-            "sandbox_id": status.sandbox_id,
-            "tool_ok": tool_result.ok,
-            "tool_output": tool_result.output,
-            "tool_error": tool_result.error,
-            "summary": summary,
-        }
-        return result
-    except Exception as exc:
-        error = str(exc)
-        raise
-    finally:
-        manager.destroy(status.sandbox_id)
-        if report_to_hub:
-            _post_gateway_event(
-                rein=rein,
-                sandbox_profile=sandbox_profile,
-                sandbox_id=status.sandbox_id,
-                project=project,
-                actor=actor,
-                title=title,
-                result=result,
-                error=error,
-            )
+    return result.model_dump(mode="json")
 
 
-def _post_gateway_event(
+def _build_result(
     *,
-    rein: Rein,
-    sandbox_profile: str,
-    sandbox_id: str,
-    project: str,
-    actor: str,
-    title: str,
-    result: dict[str, Any] | None,
+    request: ExecutionRequest,
+    request_id: str,
+    started_at: str,
+    started: float,
+    outcome: str,
+    failure_stage: str | None,
     error: str | None,
-) -> None:
-    ok = bool(result and result.get("tool_ok"))
-    hub.post_progress_event(
-        summary=f"gateway run: {title} ({'ok' if ok else 'failed'})",
-        event_type="gateway_run",
-        detail={
-            "sandbox_profile": sandbox_profile,
-            "sandbox_id": sandbox_id,
-            "rein": type(rein).__name__,
-            "project": project,
-            "actor": actor,
-            "task_title": title,
-            "ok": ok,
-            "result": result,
-            "error": error,
-        },
+    profile,
+    descriptor,
+    sandbox_id: str | None,
+    tool_result: ToolResult | None,
+    summary: ExecutionSummary | None,
+    refs: dict,
+) -> GatewayResult:
+    duration = max(0.0, time.monotonic() - started)
+    resolved_model = (
+        (summary.resolved_model if summary else None)
+        or (tool_result.resolved_model if tool_result else None)
+        or (profile.model.model if profile else None)
+    )
+    tokens_spent = (
+        summary.tokens_spent if summary and summary.tokens_spent is not None
+        else tool_result.tokens_spent if tool_result else None
+    )
+    execution_duration = (
+        summary.duration_s if summary and summary.duration_s is not None
+        else tool_result.duration_s if tool_result else None
+    )
+    evidence_error = error if failure_stage == "resolution" else (
+        f"{failure_stage} failed; inspect direct caller error" if error and failure_stage else None
+    )
+    evidence = ExecutionEvidence(
+        request_id=request_id,
+        correlation_id=request.correlation_id,
+        actor=request.actor,
+        project=request.project,
+        target_repo=request.repo,
+        contract_version=CONTRACT_VERSION,
+        profile_ref=str(profile.ref) if profile else None,
+        rein_id=descriptor.id if descriptor else None,
+        rein_version=descriptor.version if descriptor else None,
+        model_route=profile.model.route if profile else None,
+        resolved_model=resolved_model,
+        sandbox_profile=profile.sandbox_profile if profile else None,
+        sandbox_id=sandbox_id,
+        tool_profile=profile.tool_profile if profile else None,
+        outcome=outcome,
+        failure_stage=failure_stage,
+        error=evidence_error,
+        started_at=started_at,
+        finished_at=_now(),
+        duration_s=execution_duration if execution_duration is not None else duration,
+        tokens_spent=tokens_spent,
+        token_budget=profile.limits.budget_tokens if profile else None,
+        commit_sha=summary.commit_sha if summary else None,
+        artifacts=summary.artifacts if summary else [],
+        tool_events_count=len(tool_result.events) if tool_result else 0,
+        tool_events_completeness=(
+            tool_result.events_completeness if tool_result else "unavailable"
+        ),
+        refs=refs,
+    )
+    return GatewayResult(
+        ok=outcome == "succeeded",
+        evidence=evidence,
+        tool_output=tool_result.output if tool_result else "",
+        tool_error=error or (tool_result.error if tool_result else None),
+    )
+
+
+def _report(request: ExecutionRequest, result: GatewayResult) -> None:
+    if not request.report_to_hub:
+        return
+    evidence = result.evidence.model_dump(mode="json", exclude_none=True)
+    hub.post_progress_event(
+        summary=(
+            f"gateway run: {request.title} "
+            f"({'ok' if result.ok else result.evidence.outcome})"
+        ),
+        event_type="gateway_run",
+        detail=evidence,
     )

src/glas_harness/profiles.py

@@ -0,0 +1,222 @@
+"""Runtime loader for versioned Glas harness profiles and rein descriptors."""
+
+from __future__ import annotations
+
+import importlib
+import inspect
+import os
+import re
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import ValidationError
+
+from glas_harness.contract import (
+    CONTRACT_VERSION,
+    HarnessProfile,
+    Rein,
+    ReinDescriptor,
+    ResolvedExecutionContext,
+)
+
+
+class ProfileError(ValueError):
+    pass
+
+
+class UnknownProfileError(ProfileError):
+    pass
+
+
+class AmbiguousProfileError(ProfileError):
+    pass
+
+
+class IncompatibleProfileError(ProfileError):
+    pass
+
+
+_SENSITIVE_KEY = re.compile(
+    r"(^|_)(api_key|password|passwd|secret|secret_value|token_value|private_key)$",
+    re.IGNORECASE,
+)
+_SENSITIVE_VALUE = re.compile(r"^(sk-[A-Za-z0-9_-]{12,}|hvs\.[A-Za-z0-9_-]{12,})$")
+
+
+def _source_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def _default_data_dir(kind: str) -> Path:
+    env_name = "GLAS_PROFILE_DIR" if kind == "profiles" else "GLAS_REIN_REGISTRY_DIR"
+    if configured := os.environ.get(env_name):
+        return Path(configured).expanduser().resolve()
+    packaged = Path(__file__).resolve().parent / "data" / kind
+    if packaged.is_dir():
+        return packaged
+    return _source_root() / ("profiles" if kind == "profiles" else "registry/reins")
+
+
+def _read_yaml(path: Path) -> dict[str, Any]:
+    try:
+        data = yaml.safe_load(path.read_text())
+    except (OSError, yaml.YAMLError) as exc:
+        raise ProfileError(f"cannot read {path}: {exc}") from exc
+    if not isinstance(data, dict):
+        raise ProfileError(f"{path}: expected a YAML object")
+    _reject_inline_secrets(data, path=path)
+    return data
+
+
+def _reject_inline_secrets(value: Any, *, path: Path, key_path: str = "") -> None:
+    if isinstance(value, dict):
+        for key, child in value.items():
+            key_str = str(key)
+            child_path = f"{key_path}.{key_str}" if key_path else key_str
+            if _SENSITIVE_KEY.search(key_str) and child not in (None, "", [], {}):
+                raise ProfileError(f"{path}: inline secret material forbidden at {child_path}")
+            _reject_inline_secrets(child, path=path, key_path=child_path)
+    elif isinstance(value, list):
+        for index, child in enumerate(value):
+            _reject_inline_secrets(child, path=path, key_path=f"{key_path}[{index}]")
+    elif isinstance(value, str) and _SENSITIVE_VALUE.match(value):
+        raise ProfileError(f"{path}: token-looking inline value forbidden at {key_path}")
+
+
+def _capability_satisfies(actual: Any, required: Any) -> bool:
+    if isinstance(required, list):
+        if not isinstance(actual, list):
+            return False
+        return all(item in actual for item in required)
+    return actual == required
+
+
+class ProfileCatalog:
+    def __init__(
+        self,
+        profile_dir: str | Path | None = None,
+        rein_dir: str | Path | None = None,
+    ) -> None:
+        self.profile_dir = Path(profile_dir) if profile_dir else _default_data_dir("profiles")
+        self.rein_dir = Path(rein_dir) if rein_dir else _default_data_dir("reins")
+        self._profiles: dict[tuple[str, str], HarnessProfile] | None = None
+        self._reins: dict[str, ReinDescriptor] | None = None
+
+    def profiles(self) -> dict[tuple[str, str], HarnessProfile]:
+        if self._profiles is None:
+            loaded: dict[tuple[str, str], HarnessProfile] = {}
+            for path in sorted(self.profile_dir.glob("*.yaml")):
+                try:
+                    profile = HarnessProfile.model_validate(_read_yaml(path))
+                except ValidationError as exc:
+                    raise ProfileError(f"{path}: invalid harness profile: {exc}") from exc
+                key = (profile.id, profile.version)
+                if key in loaded:
+                    raise ProfileError(f"duplicate harness profile {profile.id}@{profile.version}")
+                loaded[key] = profile
+            self._profiles = loaded
+        return self._profiles
+
+    def reins(self) -> dict[str, ReinDescriptor]:
+        if self._reins is None:
+            loaded: dict[str, ReinDescriptor] = {}
+            for path in sorted(self.rein_dir.glob("*.yaml")):
+                try:
+                    descriptor = ReinDescriptor.model_validate(_read_yaml(path))
+                except ValidationError as exc:
+                    raise ProfileError(f"{path}: invalid rein descriptor: {exc}") from exc
+                if descriptor.id in loaded:
+                    raise ProfileError(f"duplicate rein descriptor {descriptor.id}")
+                loaded[descriptor.id] = descriptor
+            self._reins = loaded
+        return self._reins
+
+    def resolve(self, reference: str) -> tuple[HarnessProfile, ReinDescriptor]:
+        profile_id, separator, version = reference.partition("@")
+        candidates = [
+            profile
+            for (candidate_id, candidate_version), profile in self.profiles().items()
+            if candidate_id == profile_id and (not separator or candidate_version == version)
+        ]
+        if not candidates:
+            raise UnknownProfileError(f"unknown harness profile: {reference}")
+        if len(candidates) != 1:
+            refs = ", ".join(sorted(str(candidate.ref) for candidate in candidates))
+            raise AmbiguousProfileError(
+                f"ambiguous harness profile {reference}; pin one of: {refs}"
+            )
+        profile = candidates[0]
+        if profile.status != "enabled":
+            raise IncompatibleProfileError(f"harness profile disabled: {profile.ref}")
+        if profile.contract_version != CONTRACT_VERSION:
+            raise IncompatibleProfileError(
+                f"profile {profile.ref} requires contract {profile.contract_version}; "
+                f"gateway supports {CONTRACT_VERSION}"
+            )
+
+        descriptor = self.reins().get(profile.rein.id)
+        if descriptor is None:
+            raise IncompatibleProfileError(
+                f"profile {profile.ref} references unknown rein {profile.rein.id}"
+            )
+        if descriptor.status != "implemented":
+            raise IncompatibleProfileError(
+                f"rein {descriptor.id} is not enabled for governed execution "
+                f"(status={descriptor.status})"
+            )
+        if profile.contract_version not in descriptor.contract_versions:
+            raise IncompatibleProfileError(
+                f"rein {descriptor.id}@{descriptor.version} does not implement "
+                f"contract {profile.contract_version}"
+            )
+        for name, required in profile.rein.required_capabilities.items():
+            actual = descriptor.capabilities.get(name)
+            if not _capability_satisfies(actual, required):
+                raise IncompatibleProfileError(
+                    f"rein {descriptor.id} capability {name!r} is {actual!r}; "
+                    f"profile requires {required!r}"
+                )
+        return profile, descriptor
+
+    def resolve_context(self, reference: str) -> ResolvedExecutionContext:
+        profile, descriptor = self.resolve(reference)
+        return ResolvedExecutionContext(
+            contract_version=CONTRACT_VERSION,
+            profile=profile.ref,
+            rein_id=descriptor.id,
+            rein_version=descriptor.version,
+            sandbox_profile=profile.sandbox_profile,
+            tool_profile=profile.tool_profile,
+            model=profile.model,
+            limits=profile.limits,
+        )
+
+    def build_rein(self, profile: HarnessProfile, descriptor: ReinDescriptor) -> Rein:
+        module_name, class_name = descriptor.handler.split(":", 1)
+        module = importlib.import_module(module_name)
+        rein_type = getattr(module, class_name)
+        if not inspect.isclass(rein_type) or not issubclass(rein_type, Rein):
+            raise IncompatibleProfileError(
+                f"handler {descriptor.handler} does not implement Rein"
+            )
+        candidate_kwargs = {
+            "model": profile.model.model,
+            "tool_profile": profile.tool_profile,
+            "max_turns": profile.limits.max_turns,
+            "budget_tokens": profile.limits.budget_tokens,
+            "stream_tool_events": bool(profile.metadata.get("stream_tool_events", False)),
+        }
+        parameters = inspect.signature(rein_type).parameters
+        kwargs = {
+            name: value
+            for name, value in candidate_kwargs.items()
+            if name in parameters and value is not None
+        }
+        return rein_type(**kwargs)
+
+    def validate_all(self) -> list[ResolvedExecutionContext]:
+        return [
+            self.resolve_context(str(profile.ref))
+            for profile in sorted(self.profiles().values(), key=lambda item: (item.id, item.version))
+        ]

src/glas_harness/reins/_shared.py

@@ -26,3 +26,19 @@ def write_task_file(title: str, description: str, target_repo: str, **extra: Any
     json.dump(task_spec, fd)
     fd.close()
     return fd.name
+
+
+def parse_json_object(text: str) -> dict[str, Any]:
+    """Parse a rein CLI's final JSON object from otherwise human-readable output."""
+
+    decoder = json.JSONDecoder()
+    for index, character in enumerate(text):
+        if character != "{":
+            continue
+        try:
+            value, end = decoder.raw_decode(text[index:])
+        except json.JSONDecodeError:
+            continue
+        if isinstance(value, dict) and not text[index + end :].strip():
+            return value
+    return {}

src/glas_harness/reins/rein_aharness.py

@@ -25,8 +25,15 @@ import shutil
 import subprocess
 from typing import Any
 
-from glas_harness.contract import Rein, SandboxHandle, ToolCall, ToolResult
-from glas_harness.reins._shared import git_head, write_task_file
+from glas_harness.contract import (
+    ExecutionSummary,
+    HarnessProfile,
+    Rein,
+    SandboxHandle,
+    ToolCall,
+    ToolResult,
+)
+from glas_harness.reins._shared import git_head, parse_json_object, write_task_file
 
 
 class ReinAharnessNotInstalled(RuntimeError):
@@ -34,9 +41,20 @@ class ReinAharnessNotInstalled(RuntimeError):
 
 
 class ReinAharness(Rein):
-    def __init__(self, cli_bin: str = "rein-aharness", stream_tool_events: bool = False) -> None:
+    def __init__(
+        self,
+        cli_bin: str = "rein-aharness",
+        stream_tool_events: bool = False,
+        model: str | None = None,
+        tool_profile: str | None = None,
+        budget_tokens: int | None = None,
+    ) -> None:
         self.cli_bin = cli_bin
         self.stream_tool_events = stream_tool_events
+        self.model = model
+        self.tool_profile = tool_profile
+        self.budget_tokens = budget_tokens
+        self._last_result: dict[str, Any] = {}
 
     def _bin(self) -> str:
         resolved = shutil.which(self.cli_bin)
@@ -47,7 +65,7 @@ class ReinAharness(Rein):
         return resolved
 
     def start_session(
-        self, profile: dict[str, Any], inputs: dict[str, str], sandbox: SandboxHandle
+        self, profile: HarnessProfile, inputs: dict[str, str], sandbox: SandboxHandle
     ) -> dict[str, str]:
         target_repo = (
             inputs.get("target_repo")
@@ -58,7 +76,11 @@ class ReinAharness(Rein):
             raise ValueError("no target_repo resolvable from inputs or sandbox reachability")
 
         task_file = inputs.get("task_file") or write_task_file(
-            inputs["title"], inputs["description"], target_repo, agent=inputs.get("agent", "coach")
+            inputs["title"],
+            inputs["description"],
+            target_repo,
+            agent=inputs.get("agent", "coach"),
+            timeout_seconds=profile.limits.timeout_seconds or 600,
         )
 
         return {
@@ -79,10 +101,29 @@ class ReinAharness(Rein):
         ]
         if self.stream_tool_events:
             argv.append("--stream-tool-events")
+        if self.model:
+            argv += ["--model", self.model]
+        if self.tool_profile:
+            argv += ["--tool-profile", self.tool_profile]
+        if self.budget_tokens:
+            argv += ["--budget-tokens", str(self.budget_tokens)]
         proc = subprocess.run(argv, capture_output=True, text=True)
         ok = proc.returncode == 0
         output, events = self._split_stream_events(proc.stdout)
-        return ToolResult(ok=ok, output=output, error=None if ok else proc.stderr, events=events)
+        self._last_result = parse_json_object(output)
+        return ToolResult(
+            ok=ok,
+            output=output,
+            error=None if ok else (proc.stderr or self._last_result.get("reason")),
+            events=events,
+            events_completeness="complete" if self.stream_tool_events else "unavailable",
+            tokens_spent=self._last_result.get("tokens_spent"),
+            duration_s=self._last_result.get("execution_time_s"),
+            resolved_model=self._last_result.get("model") or self.model,
+            metadata={
+                "tool_profile": self._last_result.get("tool_profile") or self.tool_profile,
+            },
+        )
 
     @staticmethod
     def _split_stream_events(stdout: str) -> tuple[str, list[dict[str, Any]]]:
@@ -105,10 +146,25 @@ class ReinAharness(Rein):
             remaining_lines.append(line)
         return "\n".join(remaining_lines), events
 
-    def end_session(self, session: dict[str, str]) -> dict[str, str]:
+    def end_session(self, session: dict[str, str]) -> ExecutionSummary:
         head_after = git_head(session["target_repo"])
         committed = bool(head_after) and head_after != session.get("head_before")
-        return {
-            "commit_sha": head_after or "",
-            "committed": str(committed),
-        }
+        reported_ok = bool(self._last_result.get("ok", committed))
+        reason = self._last_result.get("reason") or None
+        outcome = "succeeded" if committed and reported_ok else (
+            "refused" if isinstance(reason, str) and reason.startswith("refused:") else "failed"
+        )
+        return ExecutionSummary(
+            commit_sha=head_after or None,
+            committed=committed,
+            outcome=outcome,
+            reason=reason,
+            tokens_spent=self._last_result.get("tokens_spent"),
+            duration_s=self._last_result.get("execution_time_s"),
+            resolved_model=self._last_result.get("model") or self.model,
+            artifacts=[head_after] if committed and head_after else [],
+            metadata={
+                "tool_profile": self._last_result.get("tool_profile") or self.tool_profile,
+                "persona_source": self._last_result.get("persona_source"),
+            },
+        )

src/glas_harness/reins/rein_openweights.py

@@ -13,8 +13,15 @@ import shutil
 import subprocess
 from typing import Any
 
-from glas_harness.contract import Rein, SandboxHandle, ToolCall, ToolResult
-from glas_harness.reins._shared import git_head, write_task_file
+from glas_harness.contract import (
+    ExecutionSummary,
+    HarnessProfile,
+    Rein,
+    SandboxHandle,
+    ToolCall,
+    ToolResult,
+)
+from glas_harness.reins._shared import git_head, parse_json_object, write_task_file
 
 
 class ReinOpenWeightsNotInstalled(RuntimeError):
@@ -22,9 +29,20 @@ class ReinOpenWeightsNotInstalled(RuntimeError):
 
 
 class ReinOpenWeights(Rein):
-    def __init__(self, cli_bin: str = "rein-openweights", model: str | None = None) -> None:
+    def __init__(
+        self,
+        cli_bin: str = "rein-openweights",
+        model: str | None = None,
+        max_turns: int | None = None,
+        budget_tokens: int | None = None,
+        tool_profile: str | None = None,
+    ) -> None:
         self.cli_bin = cli_bin
         self.model = model
+        self.max_turns = max_turns
+        self.budget_tokens = budget_tokens
+        self.tool_profile = tool_profile
+        self._last_result: dict[str, Any] = {}
 
     def _bin(self) -> str:
         resolved = shutil.which(self.cli_bin)
@@ -35,7 +53,7 @@ class ReinOpenWeights(Rein):
         return resolved
 
     def start_session(
-        self, profile: dict[str, Any], inputs: dict[str, str], sandbox: SandboxHandle
+        self, profile: HarnessProfile, inputs: dict[str, str], sandbox: SandboxHandle
     ) -> dict[str, str]:
         target_repo = (
             inputs.get("target_repo")
@@ -46,7 +64,10 @@ class ReinOpenWeights(Rein):
             raise ValueError("no target_repo resolvable from inputs or sandbox reachability")
 
         task_file = inputs.get("task_file") or write_task_file(
-            inputs["title"], inputs["description"], target_repo
+            inputs["title"],
+            inputs["description"],
+            target_repo,
+            timeout_seconds=profile.limits.timeout_seconds or 600,
         )
 
         return {
@@ -60,14 +81,48 @@ class ReinOpenWeights(Rein):
         argv = [self._bin(), "run", "--task-file", session["task_file"], "--no-hub"]
         if self.model:
             argv += ["--model", self.model]
+        if self.max_turns:
+            argv += ["--max-turns", str(self.max_turns)]
+        if self.budget_tokens:
+            argv += ["--budget-tokens", str(self.budget_tokens)]
+        if self.tool_profile:
+            argv += ["--tool-profile", self.tool_profile]
         proc = subprocess.run(argv, capture_output=True, text=True)
         ok = proc.returncode == 0
-        return ToolResult(ok=ok, output=proc.stdout, error=None if ok else proc.stderr)
+        self._last_result = parse_json_object(proc.stdout)
+        return ToolResult(
+            ok=ok,
+            output=proc.stdout,
+            error=None if ok else (proc.stderr or self._last_result.get("reason")),
+            events_completeness="unavailable",
+            tokens_spent=self._last_result.get("tokens_spent"),
+            duration_s=self._last_result.get("execution_time_s"),
+            resolved_model=self._last_result.get("model") or self.model,
+            metadata={
+                "turns": self._last_result.get("turns"),
+                "tool_profile": self._last_result.get("tool_profile") or self.tool_profile,
+            },
+        )
 
-    def end_session(self, session: dict[str, str]) -> dict[str, str]:
+    def end_session(self, session: dict[str, str]) -> ExecutionSummary:
         head_after = git_head(session["target_repo"])
         committed = bool(head_after) and head_after != session.get("head_before")
-        return {
-            "commit_sha": head_after or "",
-            "committed": str(committed),
-        }
+        reported_ok = bool(self._last_result.get("ok", committed))
+        reason = self._last_result.get("reason") or None
+        outcome = "succeeded" if committed and reported_ok else (
+            "refused" if reason == "no OpenRouter credential resolved" else "failed"
+        )
+        return ExecutionSummary(
+            commit_sha=head_after or None,
+            committed=committed,
+            outcome=outcome,
+            reason=reason,
+            tokens_spent=self._last_result.get("tokens_spent"),
+            duration_s=self._last_result.get("execution_time_s"),
+            resolved_model=self._last_result.get("model") or self.model,
+            artifacts=[head_after] if committed and head_after else [],
+            metadata={
+                "turns": self._last_result.get("turns"),
+                "tool_profile": self._last_result.get("tool_profile") or self.tool_profile,
+            },
+        )