From 489fd20258da32107517ef160eba330d0d6c6113 Mon Sep 17 00:00:00 2001 From: hadarishav Date: Wed, 26 Aug 2026 18:59:15 +0530 Subject: [PATCH] Harden hosted environment resolution and reporting --- src/fi/alk/harness/bundle.py | 29 +- src/fi/alk/harness/cli.py | 231 +++++++++- src/fi/alk/harness/environment_resolution.py | 442 +++++++++++++++++++ src/fi/alk/harness/executor.py | 6 +- src/fi/alk/harness/failure_reporting.py | 108 +++++ src/fi/alk/harness/job.py | 1 + src/fi/alk/harness/platform.py | 74 +++- src/fi/alk/harness/provision.py | 106 ++++- src/fi/alk/harness/run/grade.py | 52 ++- src/fi/alk/harness/sandbox_server.py | 168 ++++++- tests/test_harness.py | 96 +++- tests/test_harness_architecture.py | 136 +++++- tests/test_harness_environment_resolution.py | 168 +++++++ tests/test_harness_sandbox_server.py | 93 +++- tests/test_harness_service_environments.py | 23 + 15 files changed, 1673 insertions(+), 60 deletions(-) create mode 100644 src/fi/alk/harness/environment_resolution.py create mode 100644 src/fi/alk/harness/failure_reporting.py create mode 100644 tests/test_harness_environment_resolution.py diff --git a/src/fi/alk/harness/bundle.py b/src/fi/alk/harness/bundle.py index 38f20d02..bc8f0381 100644 --- a/src/fi/alk/harness/bundle.py +++ b/src/fi/alk/harness/bundle.py @@ -539,6 +539,21 @@ def export_session_bundle( create_environment_plan, write_environment_plan, ) + from .environment_resolution import ( + ENVIRONMENT_RESOLUTION_FILE, + load_environment_resolution, + ) + + resolution = None + if (session / ENVIRONMENT_RESOLUTION_FILE).is_file(): + resolution = load_environment_resolution(session) + if resolution.source_fingerprint != raw["provenance"]["source_digest"]: + raise BundleError("environment_resolution_source_fingerprint_mismatch") + shutil.copy2( + session / ENVIRONMENT_RESOLUTION_FILE, + staging / ENVIRONMENT_RESOLUTION_FILE, + ) + generated.append(ENVIRONMENT_RESOLUTION_FILE) plan = create_environment_plan( source=PlanSource( @@ -553,6 +568,15 @@ def export_session_bundle( readiness=[ReadinessProbe.model_validate(item) for item in readiness], metadata={ "managed": bool(provisioned and provisioned.managed), + "packaging_type": ( + provisioned.packaging_type if provisioned else "unknown" + ), + "runtime_adapter": ( + provisioned.runtime_adapter if provisioned else "unknown" + ), + "selected_runtime": ( + provisioned.selected_runtime if provisioned else "" + ), "generated_runtime_fingerprint": ( provisioned.runtime_fingerprint if provisioned else "" ), @@ -561,7 +585,10 @@ def export_session_bundle( write_environment_plan(staging, plan) generated.append(ENVIRONMENT_PLAN_FILE) raw["provenance"]["generated_files"] = generated - raw["metadata"] = {"environment_plan_digest": plan.digest} + raw["metadata"] = { + "environment_plan_digest": plan.digest, + "environment_resolution_digest": resolution.digest if resolution else "", + } seal_bundle(staging, raw) if final.exists(): shutil.rmtree(final) diff --git a/src/fi/alk/harness/cli.py b/src/fi/alk/harness/cli.py index d7f7ecaf..a7963d7a 100644 --- a/src/fi/alk/harness/cli.py +++ b/src/fi/alk/harness/cli.py @@ -10,6 +10,7 @@ import argparse import asyncio +from contextlib import suppress import json import os import sys @@ -39,6 +40,92 @@ from .world.snapshot import saved as world_saved +def _environment_failure_context( + source: str | Path, contract: Any | None +) -> dict[str, str]: + """Describe the selected packaging lane without exposing runner-local paths.""" + from .provision import environment_adapter_context + + return environment_adapter_context(source, contract) + + +def _relevant_source_file_count(source: Path) -> int: + ignored = {".git", ".venv", "venv", "node_modules", "dist", "build", "__pycache__"} + count = 0 + for path in source.rglob("*"): + if any(part in ignored for part in path.parts) or not path.is_file(): + continue + count += 1 + if count >= 10_000: + break + return count + + +def _stage_progress( + label: str, stage_args: argparse.Namespace, elapsed: int +) -> dict[str, Any]: + destination = Path(stage_args.out) + if label == "understand": + total = _relevant_source_file_count(Path(stage_args.path)) + return { + "detail": f"Understanding source · inspecting {total} relevant files", + "activity": "source_inspection", + "total_files": total, + "elapsed_seconds": elapsed, + } + if label == "environment": + ready = (destination / "environment.json").is_file() + return { + "detail": ( + "Building environment · validating runtime and seed data" + if ready + else "Building environment · resolving runtime and installing dependencies" + ), + "activity": "environment_validation" if ready else "environment_build", + "elapsed_seconds": elapsed, + } + if label == "scenarios": + created = 0 + try: + value = json.loads( + (destination / "scenarios.json").read_text(encoding="utf-8") + ) + created = len(value) if isinstance(value, list) else 0 + except (OSError, ValueError): + pass + wanted = int(getattr(stage_args, "count", 0) or 0) + return { + "detail": ( + f"Generating scenarios · validating {created}/{wanted}" + if created + else f"Generating scenarios · creating {wanted} scenarios" + ), + "activity": "scenario_validation" if created else "scenario_generation", + "completed": created, + "total": wanted, + "elapsed_seconds": elapsed, + } + completed = 0 + runs = destination / "runs" + if runs.is_dir(): + completed = sum( + 1 for path in runs.glob("run-*/*/result.json") if path.is_file() + ) + return { + "detail": f"Running scenarios · {completed} results committed", + "activity": "scenario_execution", + "completed": completed, + "elapsed_seconds": elapsed, + } + + +def _progress_interval_seconds() -> float: + try: + return max(1.0, float(os.getenv("ALK_HARNESS_PROGRESS_INTERVAL_SECONDS", "15"))) + except ValueError: + return 15.0 + + def _source_root(destination: Path, explicit: str = "") -> str: """Recover the source path for commands resumed from a session folder.""" if explicit.strip(): @@ -106,6 +193,8 @@ def _guidance(args: argparse.Namespace) -> str: async def _understand(args: argparse.Namespace) -> int: + from .failure_reporting import record_stage_failure + source = resolve(args.kind, name=args.name, root=args.path) stage, destination = open_stage( source, @@ -132,6 +221,11 @@ async def _understand(args: argparse.Namespace) -> int: contract = load(destination) if contract is None: + record_stage_failure( + "understanding_contract_missing", + "Agent understanding finished without producing a valid contract.", + source=args.path, + ) print("\nNo contract was submitted.", file=sys.stderr) return 1 print( @@ -178,10 +272,14 @@ async def _converse( async def _build(args: argparse.Namespace) -> int: + from .failure_reporting import record_stage_failure + destination = Path(args.out) if args.out else artifact_dir(args.name) contract = load(destination) if contract is None: - print(f"No contract at {destination}. Run `understand` first.", file=sys.stderr) + message = "No saved agent contract is available. Run understanding first." + record_stage_failure("agent_contract_missing", message) + print(message, file=sys.stderr) return 1 print(f"agent: {contract.agent} ({len(contract.tools)} tools)") @@ -192,6 +290,12 @@ async def _build(args: argparse.Namespace) -> int: try: require_buildable(contract, source_root) except RuntimeError as failed: + record_stage_failure( + "environment_not_buildable", + failed, + source=source_root, + details=_environment_failure_context(source_root, contract), + ) print(str(failed), file=sys.stderr) return 1 @@ -202,6 +306,12 @@ async def _build(args: argparse.Namespace) -> int: provision_if_present, source_root, destination, contract ) except ProvisionError as failed: + record_stage_failure( + "environment_provision_failed", + f"Cannot create the source environment: {failed}", + source=source_root, + details=_environment_failure_context(source_root, contract), + ) print(f"Cannot create the source environment: {failed}", file=sys.stderr) return 1 if environment is not None: @@ -227,6 +337,12 @@ async def _build(args: argparse.Namespace) -> int: ) if not world_saved(destination): + record_stage_failure( + "environment_world_missing", + "Environment generation finished without saving a validated world.", + source=source_root, + details=_environment_failure_context(source_root, contract), + ) print("\nNo world was saved.", file=sys.stderr) return 1 # Seal the exact environment now that its generated world exists. Local and hosted @@ -242,6 +358,12 @@ async def _build(args: argparse.Namespace) -> int: name=f"{contract.agent}-environment", ) except BundleError as failed: + record_stage_failure( + "environment_bundle_failed", + f"Cannot seal the environment bundle: {failed}", + source=source_root, + details=_environment_failure_context(source_root, contract), + ) print(f"Cannot seal the environment bundle: {failed}", file=sys.stderr) return 1 print(f"\nworld: {destination}") @@ -261,6 +383,7 @@ async def _environment(args: argparse.Namespace) -> int: ) destination = Path(args.out) + source_path = str(getattr(args, "path", "") or "") try: if args.action == "down": if not stop(destination): @@ -294,10 +417,22 @@ async def _environment(args: argparse.Namespace) -> int: if (bundle_root / ENVIRONMENT_PLAN_FILE).is_file(): load_environment_plan(bundle_root, bundle=bundle) except (BundleError, EnvironmentPlanError) as failed: + from .failure_reporting import record_stage_failure + + record_stage_failure( + "environment_bundle_invalid", + f"Environment bundle failed verification: {failed}", + ) print(f"Environment bundle failed: {failed}", file=sys.stderr) return 1 bundled_source = bundle_root / "services" / "source" if not bundled_source.is_dir(): + from .failure_reporting import record_stage_failure + + record_stage_failure( + "environment_bundle_source_missing", + "The saved environment bundle has no source snapshot.", + ) print( f"Environment bundle has no source snapshot: {bundled_source}", file=sys.stderr, @@ -311,6 +446,23 @@ async def _environment(args: argparse.Namespace) -> int: # and report that a previously valid Compose environment cannot be started. environment = provision(source_path, destination, load(destination)) except ProvisionError as failed: + from .failure_reporting import record_stage_failure + + contract = load(destination) + record_stage_failure( + failed.code, + f"Environment failed: {failed}", + source=source_path or None, + action=failed.action, + details={ + **( + _environment_failure_context(source_path, contract) + if source_path and Path(source_path).exists() + else {"failed_adapter": "saved_environment_replay"} + ), + **failed.details, + }, + ) print(f"Environment failed: {failed}", file=sys.stderr) return 1 @@ -323,12 +475,22 @@ async def _environment(args: argparse.Namespace) -> int: async def _scenarios(args: argparse.Namespace) -> int: + from .failure_reporting import record_stage_failure + destination = Path(args.out) if args.out else artifact_dir(args.name) contract = load(destination) if contract is None: + record_stage_failure( + "scenario_contract_missing", + "Scenario generation cannot start because the agent contract is missing.", + ) print(f"No contract at {destination}. Run `understand` first.", file=sys.stderr) return 1 if not world_saved(destination): + record_stage_failure( + "scenario_world_missing", + "Scenario generation cannot start because the validated environment world is missing.", + ) print(f"No world at {destination}. Run `build` first.", file=sys.stderr) return 1 @@ -363,6 +525,10 @@ async def _scenarios(args: argparse.Namespace) -> int: written = load_written(destination) if not written: + record_stage_failure( + "scenarios_not_saved", + "Scenario generation finished without producing validated scenarios.", + ) print("\nNo scenarios were saved.", file=sys.stderr) return 1 print(f"\nscenarios: {len(written)} in {destination / 'scenarios.json'}") @@ -471,9 +637,15 @@ async def _simulate(args: argparse.Namespace) -> int: from .world.snapshot import require_source_implementation destination = Path(args.out) if args.out else artifact_dir(args.name) + from .failure_reporting import record_stage_failure + contract = load(destination) written = load_written(destination) if contract is None or not written: + record_stage_failure( + "simulation_inputs_missing", + "Simulation requires a saved agent contract and validated scenarios.", + ) print( f"Need a contract and scenarios at {destination}.", file=sys.stderr, @@ -482,10 +654,15 @@ async def _simulate(args: argparse.Namespace) -> int: try: require_source_implementation(destination) except (FileNotFoundError, RuntimeError) as failed: + record_stage_failure("simulation_world_invalid", failed) print(str(failed), file=sys.stderr) return 1 chosen = [s for s in written if s.name in args.only] if args.only else written if not chosen: + record_stage_failure( + "simulation_scenarios_missing", + f"No scenario matched the requested selection: {args.only}", + ) print(f"No scenario matching {args.only}.", file=sys.stderr) return 1 @@ -560,6 +737,11 @@ def show_started(scenario: Any) -> None: # means every scenario ran and the submitted agent failed one or more checks. Hosted # execution retries/classifies the former and preserves the latter as valid RL evidence. if summary.get("unrunnable"): + record_stage_failure( + "simulation_unrunnable", + "One or more scenarios could not establish a runnable agent connection.", + details={"unrunnable_scenarios": int(summary.get("unrunnable") or 0)}, + ) return 1 return 0 if summary["passed"] == summary["scenarios"] else 2 @@ -762,6 +944,7 @@ def emit(event_type: str, stage: str, **payload: Any) -> None: ), ), ] + from .failure_reporting import clear_stage_failure, take_stage_failure from .provision import ProvisionError, stop cleanup_failed: ProvisionError | None = None @@ -778,15 +961,55 @@ def emit(event_type: str, stage: str, **payload: Any) -> None: label, operation, stage_args = stages[stage_index] print(f"\n=== {label} ===", flush=True) emit("harness.stage.started", label) - status = await operation(stage_args) + clear_stage_failure() + started = time.monotonic() + initial_progress = _stage_progress(label, stage_args, 0) + emit("harness.stage.progress", label, **initial_progress) + + async def heartbeat() -> None: + interval = _progress_interval_seconds() + while True: + await asyncio.sleep(interval) + progress = ( + {**initial_progress} + if label == "understand" + else _stage_progress( + label, stage_args, int(time.monotonic() - started) + ) + ) + progress["elapsed_seconds"] = int(time.monotonic() - started) + emit( + "harness.stage.progress", + label, + **progress, + ) + + heartbeat_task = asyncio.create_task(heartbeat()) + try: + status = await operation(stage_args) + finally: + heartbeat_task.cancel() + with suppress(asyncio.CancelledError): + await heartbeat_task + stage_failure = take_stage_failure() # A completed call suite returns 2 when the submitted agent fails one or more checks. # That is a valid RL result. Earlier stages returning non-zero are harness failures. if status and label != "calls": - emit("harness.stage.failed", label, status=status) + emit( + "harness.stage.failed", + label, + status=status, + **(stage_failure or {}), + ) print(f"\nautomatic run stopped: {label} failed", file=sys.stderr) return status if label == "calls" and status not in (0, 2): - emit("harness.stage.failed", label, status=status) + emit( + "harness.stage.failed", + label, + status=status, + **(stage_failure or {}), + ) return status emit("harness.stage.completed", label, status=status) for adjustment_id in applying[label]: diff --git a/src/fi/alk/harness/environment_resolution.py b/src/fi/alk/harness/environment_resolution.py new file mode 100644 index 00000000..2e895f28 --- /dev/null +++ b/src/fi/alk/harness/environment_resolution.py @@ -0,0 +1,442 @@ +"""One deterministic admission decision shared by preflight and provisioning. + +This is deliberately separate from :mod:`environment_plan`, which is the sealed description of +an already provisioned bundle. A resolved plan is produced before Docker is touched and records +which adapter is allowed to create that bundle. +""" + +from __future__ import annotations + +import hashlib +import json +from importlib.metadata import PackageNotFoundError, version +from pathlib import Path +from typing import Any, Literal + +from pydantic import BaseModel, Field, model_validator + +from .generated_runtime import GeneratedRuntimeError, detect_generated_runtime +from .packaging import ( + PackagingCandidate, + PackagingKind, + PackagingManifest, + inspect_packaging, +) + + +ENVIRONMENT_RESOLUTION_SCHEMA_VERSION = "futureagi.environment-resolution.v1" +ENVIRONMENT_RESOLUTION_FILE = "environment-resolution.json" + + +class DependencyDecision(BaseModel): + name: str + engine: str = "" + ownership: Literal[ + "submitted_compose", + "managed_service", + "agent_runtime", + "external_provider", + "embedded", + "unsupported", + ] + + +class ResolvedEnvironmentPlan(BaseModel): + schema_version: str = ENVIRONMENT_RESOLUTION_SCHEMA_VERSION + digest: str + source_fingerprint: str = "" + contract_digest: str = "" + harness_version: str + packaging_type: Literal["compose", "dockerfile", "unpackaged"] + runtime_adapter: Literal[ + "submitted_compose", + "managed_compose_for_dockerfile", + "generated_runtime", + "unsupported", + ] + selected_runtime: str = "" + component_root: str = "." + build_context: str = "." + dependencies: list[DependencyDecision] = Field(default_factory=list) + required_credentials: list[str] = Field(default_factory=list) + supported: bool + execution_ready: bool + code: str = "" + message: str = "" + action: str = "" + + @model_validator(mode="after") + def _consistent(self) -> "ResolvedEnvironmentPlan": + if self.supported == (self.runtime_adapter == "unsupported"): + raise ValueError("environment_resolution_support_adapter_mismatch") + names = [item.name for item in self.dependencies] + if len(names) != len(set(names)): + raise ValueError("environment_resolution_dependency_ownership_not_unique") + if self.supported and any( + item.ownership == "unsupported" for item in self.dependencies + ): + raise ValueError("environment_resolution_unsupported_dependency") + if self.required_credentials != sorted(set(self.required_credentials)): + raise ValueError("environment_resolution_credentials_not_canonical") + if _resolution_digest(self.model_dump(mode="json")) != self.digest: + raise ValueError("environment_resolution_digest_mismatch") + return self + + +def _canonical_digest(value: Any) -> str: + if hasattr(value, "model_dump"): + value = value.model_dump(mode="json") + elif hasattr(value, "__dict__"): + value = { + key: _jsonable(item) + for key, item in vars(value).items() + if not key.startswith("_") + } + encoded = json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=False + ).encode("utf-8") + return "sha256:" + hashlib.sha256(encoded).hexdigest() + + +def _jsonable(value: Any) -> Any: + if hasattr(value, "model_dump"): + return value.model_dump(mode="json") + if hasattr(value, "__dict__"): + return { + key: _jsonable(item) + for key, item in vars(value).items() + if not key.startswith("_") + } + if isinstance(value, dict): + return {str(key): _jsonable(item) for key, item in value.items()} + if isinstance(value, (list, tuple, set)): + return [_jsonable(item) for item in value] + return value + + +def _resolution_digest(raw: dict[str, Any]) -> str: + canonical = dict(raw) + canonical.pop("digest", None) + return _canonical_digest(canonical) + + +def _harness_version() -> str: + try: + return version("futureagi") + except PackageNotFoundError: + return "source-checkout" + + +def _candidate( + packaging: PackagingManifest, kind: PackagingKind, path: str +) -> PackagingCandidate | None: + normalized = Path(path).as_posix() + return next( + ( + item + for item in packaging.candidates + if item.kind is kind and item.path == normalized + ), + None, + ) + + +def _blocking( + candidate: PackagingCandidate | None, *, contract: Any | None +) -> list[str]: + if candidate is None: + return [] + return [ + finding.message + for finding in candidate.findings + if finding.blocking + and not (finding.code == "compose_env_file_missing" and contract is not None) + ] + + +def _dependency_decisions( + contract: Any | None, packaging_type: str +) -> list[DependencyDecision]: + decisions: list[DependencyDecision] = [] + managed = { + "clickhouse", + "postgres", + "mysql", + "redis", + "mongodb", + "mongo", + "qdrant", + "rabbitmq", + "nats", + "minio", + } + providers = { + "openai", + "anthropic", + "gemini", + "vertex", + "deepgram", + "cartesia", + "elevenlabs", + "livekit", + "retell", + "vapi", + "daily", + } + for index, dependency in enumerate( + list(getattr(contract, "dependencies", None) or []) + ): + name = str(getattr(dependency, "name", "") or f"dependency-{index + 1}") + engine = str(getattr(dependency, "engine", "") or "").lower() + reached = getattr(dependency, "reached", None) + external_configuration = bool( + reached + and any( + str(getattr(reached, field, "") or "").strip() + for field in ("dsn_env", "config_key", "password_from") + ) + and not str(getattr(reached, "database", "") or "").strip() + ) + description = ( + " ".join( + str(getattr(dependency, field, "") or "") + for field in ("name", "engine", "kind", "what") + ) + .lower() + .replace("-", "_") + ) + if packaging_type == "compose": + ownership = "submitted_compose" + elif any(item in description for item in managed): + ownership = "managed_service" + elif any(item in description for item in providers): + ownership = "external_provider" + elif external_configuration: + # A credential/configuration seam without a database identifies a customer-owned + # remote API. ALK supplies the reference when configured; it must not try to create + # that SaaS service or reject an agent that has its own in-process fallback. + ownership = "external_provider" + elif reached and ( + str(getattr(reached, "loader_module", "") or "") + or str(getattr(reached, "loader_function", "") or "") + or any( + marker in description + for marker in ( + "sqlite", + "in_process", + "in_memory", + "filesystem", + "local_model", + ) + ) + ): + ownership = "embedded" + elif packaging_type == "dockerfile" and engine: + ownership = "agent_runtime" + else: + ownership = "unsupported" + decisions.append( + DependencyDecision(name=name, engine=engine, ownership=ownership) + ) + return decisions + + +def _credential_names(contract: Any | None) -> list[str]: + names: set[str] = set() + store = getattr(contract, "data_store", None) + values: list[Any] = [store] + values.extend( + getattr(item, "reached", None) + for item in list(getattr(contract, "dependencies", None) or []) + ) + for value in values: + if value is None: + continue + for field in ("dsn_env", "config_key", "password_from"): + candidate = str(getattr(value, field, "") or "").strip() + if ( + candidate + and candidate.replace("_", "").isalnum() + and candidate.upper() == candidate + ): + names.add(candidate) + return sorted(names) + + +def resolve_environment_plan( + source: str | Path, + packaging: PackagingManifest | None = None, + contract: Any | None = None, + *, + source_fingerprint: str = "", +) -> ResolvedEnvironmentPlan: + """Resolve exactly one provisioner without executing source or Docker.""" + root = Path(source).expanduser().resolve() + packaging = packaging or inspect_packaging(root) + runtime = getattr(contract, "runtime", None) + explicit_compose = str(getattr(runtime, "compose_file", "") or "") + explicit_dockerfile = str(getattr(runtime, "dockerfile", "") or "") + selected = "" + packaging_type = "unpackaged" + adapter = "generated_runtime" + component = str(getattr(runtime, "workdir", "") or ".") + build_context = component + supported = True + execution_ready = True + code = message = action = "" + + if explicit_compose: + candidate = _candidate(packaging, PackagingKind.COMPOSE, explicit_compose) + selected = Path(explicit_compose).as_posix() + packaging_type, adapter = "compose", "submitted_compose" + if candidate is None: + supported = False + code = "runtime_compose_not_admitted" + message = f"Runtime Compose file is not an admitted repository candidate: {selected}" + action = "Select a discovered Compose file within the submitted repository." + elif problems := _blocking(candidate, contract=contract): + supported = False + code = "packaging_preflight_failed" + message = "; ".join(problems) + action = "Resolve the reported Compose admission findings and retry." + elif explicit_dockerfile: + candidate = _candidate(packaging, PackagingKind.DOCKERFILE, explicit_dockerfile) + selected = Path(explicit_dockerfile).as_posix() + packaging_type, adapter = "dockerfile", "managed_compose_for_dockerfile" + if not (root / selected).is_file(): + supported = False + code = "runtime_dockerfile_missing" + message = f"Runtime Dockerfile does not exist: {selected}" + action = "Provide the Dockerfile in the submitted repository or correct runtime.dockerfile." + elif problems := _blocking(candidate, contract=contract): + supported = False + code = "packaging_preflight_failed" + message = "; ".join(problems) + action = "Resolve the reported Dockerfile admission findings and retry." + elif packaging.selected_kind is PackagingKind.COMPOSE and packaging.selected_path: + selected = packaging.selected_path + packaging_type, adapter = "compose", "submitted_compose" + if contract is None and not packaging.agent_runtime_packaged: + execution_ready = False + code = "compose_agent_runtime_missing" + message = "Compose describes infrastructure but no packaged agent runtime." + action = "Package the agent runtime or submit a source that can use the generated runtime adapter." + elif ( + packaging.selected_kind is PackagingKind.DOCKERFILE and packaging.selected_path + ): + selected = packaging.selected_path + packaging_type, adapter = "dockerfile", "managed_compose_for_dockerfile" + elif packaging.candidates: + supported = False + adapter = "unsupported" + code = "packaging_selection_required" + details = list(packaging.notes) + details.extend( + finding.message + for candidate in packaging.candidates + for finding in candidate.findings + if finding.blocking + ) + message = ( + "; ".join(details) or "Repository packaging is ambiguous or unsupported." + ) + action = "Select one runnable component/build context or fix the blocking packaging findings." + else: + try: + generated = detect_generated_runtime(root, runtime) + selected = generated.dependency_file + component = generated.component or "." + build_context = component + except GeneratedRuntimeError as exc: + if contract is None: + # Understanding may still prove a remote/in-process agent. Submission is allowed, + # but the same resolver is called again with that contract before provisioning. + selected = "pending-contract" + else: + supported = False + adapter = "unsupported" + code = str(exc).split(":", 1)[0] + managed_dependencies = [ + item.engine + for item in _dependency_decisions(contract, "unpackaged") + if item.ownership == "managed_service" and item.engine + ] + message = ( + "the agent requires " + + ", ".join(managed_dependencies) + + " but ships neither Compose nor a Dockerfile; " + + str(exc) + if managed_dependencies + else str(exc) + ) + action = "Add supported packaging or declare an unambiguous Python/Node runtime." + + dependencies = _dependency_decisions(contract, packaging_type) + unsupported = [ + item.name for item in dependencies if item.ownership == "unsupported" + ] + if supported and unsupported: + supported = False + adapter = "unsupported" + code = "unsupported_dependency" + message = "No environment owner could be determined for: " + ", ".join( + unsupported + ) + action = "Package the dependency in Compose or expose a supported configuration seam." + if not supported: + execution_ready = False + + raw: dict[str, Any] = { + "schema_version": ENVIRONMENT_RESOLUTION_SCHEMA_VERSION, + "digest": "sha256:" + "0" * 64, + "source_fingerprint": source_fingerprint, + "contract_digest": _canonical_digest(contract) if contract is not None else "", + "harness_version": _harness_version(), + "packaging_type": packaging_type, + "runtime_adapter": adapter, + "selected_runtime": selected or "automatic", + "component_root": component or ".", + "build_context": build_context or ".", + "dependencies": [item.model_dump(mode="json") for item in dependencies], + "required_credentials": _credential_names(contract), + "supported": supported, + "execution_ready": execution_ready, + "code": code, + "message": message, + "action": action, + } + raw["digest"] = _resolution_digest(raw) + return ResolvedEnvironmentPlan.model_validate(raw) + + +def write_environment_resolution( + root: str | Path, plan: ResolvedEnvironmentPlan +) -> Path: + target = Path(root) / ENVIRONMENT_RESOLUTION_FILE + target.parent.mkdir(parents=True, exist_ok=True) + temporary = target.with_name(f".{target.name}.tmp") + temporary.write_text(plan.model_dump_json(indent=2) + "\n", encoding="utf-8") + temporary.replace(target) + return target + + +def load_environment_resolution(root: str | Path) -> ResolvedEnvironmentPlan: + target = Path(root) / ENVIRONMENT_RESOLUTION_FILE + if not target.is_file(): + raise ValueError(f"environment_resolution_missing: {target}") + try: + return ResolvedEnvironmentPlan.model_validate_json( + target.read_text(encoding="utf-8") + ) + except (OSError, ValueError) as exc: + raise ValueError(f"environment_resolution_invalid: {exc}") from exc + + +__all__ = [ + "DependencyDecision", + "ENVIRONMENT_RESOLUTION_FILE", + "ResolvedEnvironmentPlan", + "load_environment_resolution", + "resolve_environment_plan", + "write_environment_resolution", +] diff --git a/src/fi/alk/harness/executor.py b/src/fi/alk/harness/executor.py index d4d8fc4c..6184810a 100644 --- a/src/fi/alk/harness/executor.py +++ b/src/fi/alk/harness/executor.py @@ -283,6 +283,9 @@ def _failure_from_events(output: Path) -> HarnessFailure: FailureDomain.ARTIFACT, ), }.get(label, (HarnessStage.RUNNING, FailureDomain.INFRASTRUCTURE)) + event_details = failed.get("details") + details = dict(event_details) if isinstance(event_details, dict) else {} + details["status"] = failed.get("status", 1) return HarnessFailure( domain=domain, stage=stage, @@ -291,7 +294,8 @@ def _failure_from_events(output: Path) -> HarnessFailure: # A job replay can repeat real calls. Only a failing stage that explicitly proves # it is safe may opt in to retry; deterministic agent/grading failures never do. retryable=bool(failed.get("retryable", False)), - details={"status": failed.get("status", 1)}, + details=details, + action=str(failed.get("action") or ""), ) diff --git a/src/fi/alk/harness/failure_reporting.py b/src/fi/alk/harness/failure_reporting.py new file mode 100644 index 00000000..3f022380 --- /dev/null +++ b/src/fi/alk/harness/failure_reporting.py @@ -0,0 +1,108 @@ +"""Safe, structured failure handoff between CLI stages and harness controllers.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path +import re +from typing import Any + + +FAILURE_PATH_ENVIRONMENT = "ALK_HARNESS_FAILURE_PATH" +_last_failure: dict[str, Any] | None = None +_SENSITIVE_ASSIGNMENT = re.compile( + r"(?i)\b(api[_-]?key|secret|token|password|authorization|credential)" + r"\s*[:=]\s*([^\s,;]+)" +) + + +def sanitize_failure_message( + message: object, *, source: str | Path | None = None +) -> str: + """Return a bounded explanation suitable for a customer-visible job result.""" + value = str(message).replace("\r", " ").replace("\n", " ").strip() + if source: + source_value = str(Path(source).expanduser().resolve()) + value = value.replace(source_value, "the submitted repository") + for name, secret in os.environ.items(): + lowered = name.lower() + if ( + secret + and len(secret) >= 6 + and any( + marker in lowered + for marker in ("secret", "token", "password", "api_key", "credential") + ) + ): + value = value.replace(secret, "[REDACTED]") + value = _SENSITIVE_ASSIGNMENT.sub( + lambda match: f"{match.group(1)}=[REDACTED]", value + ) + return value[:1000] or "The harness stage failed without an explanation" + + +def clear_stage_failure() -> None: + global _last_failure + _last_failure = None + configured = os.getenv(FAILURE_PATH_ENVIRONMENT, "").strip() + if configured: + Path(configured).unlink(missing_ok=True) + + +def record_stage_failure( + code: str, + message: object, + *, + source: str | Path | None = None, + details: dict[str, Any] | None = None, + retryable: bool = False, + action: str = "", +) -> dict[str, Any]: + """Record one sanitized failure in memory and, when requested, a private sidecar.""" + global _last_failure + safe_details = { + str(key): value + for key, value in (details or {}).items() + if isinstance(value, (str, int, float, bool)) or value is None + } + _last_failure = { + "code": re.sub(r"[^a-z0-9_]+", "_", code.lower()).strip("_") or "stage_failed", + "detail": sanitize_failure_message(message, source=source), + "details": safe_details, + "retryable": bool(retryable), + "action": sanitize_failure_message(action, source=source) if action else "", + } + configured = os.getenv(FAILURE_PATH_ENVIRONMENT, "").strip() + if configured: + target = Path(configured) + target.parent.mkdir(parents=True, exist_ok=True) + temporary = target.with_name(f".{target.name}.tmp") + temporary.write_text(json.dumps(_last_failure), encoding="utf-8") + temporary.replace(target) + return dict(_last_failure) + + +def take_stage_failure() -> dict[str, Any] | None: + global _last_failure + value = _last_failure + _last_failure = None + return dict(value) if value else None + + +def load_stage_failure(path: str | Path) -> dict[str, Any] | None: + try: + value = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + return value if isinstance(value, dict) else None + + +__all__ = [ + "FAILURE_PATH_ENVIRONMENT", + "clear_stage_failure", + "load_stage_failure", + "record_stage_failure", + "sanitize_failure_message", + "take_stage_failure", +] diff --git a/src/fi/alk/harness/job.py b/src/fi/alk/harness/job.py index 794b17b5..4ba43944 100644 --- a/src/fi/alk/harness/job.py +++ b/src/fi/alk/harness/job.py @@ -241,6 +241,7 @@ class HarnessFailure(BaseModel): message: str retryable: bool = False details: dict[str, JsonValue] = Field(default_factory=dict) + action: str = "" @property def owner(self) -> FailureOwner: diff --git a/src/fi/alk/harness/platform.py b/src/fi/alk/harness/platform.py index 3d8a4169..f6ce0620 100644 --- a/src/fi/alk/harness/platform.py +++ b/src/fi/alk/harness/platform.py @@ -38,6 +38,7 @@ BASE_URL = ("HARNESS_PLATFORM_URL", "FI_BASE_URL") API_KEY = ("HARNESS_PLATFORM_API_KEY", "FI_API_KEY") SECRET_KEY = ("HARNESS_PLATFORM_SECRET_KEY", "FI_SECRET_KEY") +WORKSPACE_ID = "HARNESS_PLATFORM_WORKSPACE_ID" def _setting(names: tuple[str, ...]) -> str: @@ -141,17 +142,23 @@ def __init__(self, base: str = "", key: str = "", secret: str = "") -> None: self.key = key or _setting(API_KEY) self.secret = secret or _setting(SECRET_KEY) + def _headers(self) -> dict[str, str]: + headers = { + "X-Api-Key": self.key, + "X-Secret-Key": self.secret, + } + workspace_id = os.environ.get(WORKSPACE_ID, "").strip() + if workspace_id: + headers["X-Workspace-Id"] = workspace_id + return headers + def _call( self, path: str, payload: dict[str, Any], method: str = "POST" ) -> dict[str, Any]: request = urllib.request.Request( f"{self.base}{INGESTION}{path}", data=json.dumps(payload).encode(), - headers={ - "Content-Type": "application/json", - "X-Api-Key": self.key, - "X-Secret-Key": self.secret, - }, + headers={"Content-Type": "application/json", **self._headers()}, method=method, ) try: @@ -242,8 +249,7 @@ def recording(self, call_execution_id: str, audio: Path) -> dict[str, Any]: data=body, headers={ "Content-Type": f"multipart/form-data; boundary={edge}", - "X-Api-Key": self.key, - "X-Secret-Key": self.secret, + **self._headers(), }, ) try: @@ -353,8 +359,7 @@ def evaluations_of(result: Any) -> list[dict[str, Any]]: # attach the already-computed result to that template/config instead of # merely leaving a second, disconnected EvalTemplate in the library. platform_template = decided_by.rsplit(" (", 1)[0] if decided_by else "" - judged.append( - { + evaluation = { "name": getattr(check, "name", ""), "kind": getattr(check, "kind", "") or "checkpoint", "passed": bool(getattr(check, "passed", False)), @@ -362,7 +367,9 @@ def evaluations_of(result: Any) -> list[dict[str, Any]]: "decided_by": decided_by[:2000], "platform_template": platform_template[:2000], } - ) + if getattr(check, "grading_error", False): + evaluation["grading_error"] = True + judged.append(evaluation) for metric in (getattr(result, "measured", None) or {}).get("metrics") or []: if not metric.get("applicable", True): continue @@ -377,6 +384,30 @@ def evaluations_of(result: Any) -> list[dict[str, Any]]: return [one for one in judged if one["name"]] +def conversation_seconds(result: Any) -> int: + """Return user-visible conversation time, excluding setup and retry waits.""" + exchanges = getattr(result, "exchanges", None) or [] + starts = [ + float(exchange["start_time_ms"]) + for exchange in exchanges + if isinstance(exchange, dict) + and isinstance(exchange.get("start_time_ms"), (int, float)) + ] + ends = [ + float(exchange["end_time_ms"]) + for exchange in exchanges + if isinstance(exchange, dict) + and isinstance(exchange.get("end_time_ms"), (int, float)) + ] + if starts and ends: + return max(0, int((max(ends) - min(starts)) / 1000)) + if not exchanges and getattr(result, "problems", None): + return 0 + # Text and older connector results may not carry media timestamps. Their + # measured scenario time remains the best available conversation duration. + return max(0, int(getattr(result, "seconds", 0) or 0)) + + def result_of(result: Any) -> dict[str, Any]: """One scenario's outcome, in the shape the ingestion API takes. @@ -393,11 +424,26 @@ def result_of(result: Any) -> dict[str, Any]: for check in getattr(result, "checkpoints", None) or [] ] problems = list(getattr(result, "problems", None) or []) + grading_failures = list(getattr(result, "grading_failures", None) or []) + evaluations = evaluations_of(result) + evaluation_coverage = { + "expected": len(evaluations), + "executed": sum( + 1 for evaluation in evaluations if not evaluation.get("grading_error") + ), + "failed": sum( + 1 for evaluation in evaluations if evaluation.get("grading_error") + ), + "complete": not grading_failures, + } payload: dict[str, Any] = { # A scenario that never ran is not a scenario the agent failed, and the two must not # arrive as the same status. + # Call status is transport/execution lifecycle only. An unavailable + # evaluator is a grading failure and travels in coverage metadata; it + # must not rewrite a successfully completed call to FAILED. "status": "failed" if problems else "completed", - "duration_seconds": max(0, int(getattr(result, "seconds", 0) or 0)), + "duration_seconds": conversation_seconds(result), "ended_reason": (getattr(result, "ended", "") or "")[:10000], "call_summary": (getattr(result, "line", lambda: "")() or "")[:2000], "transcript": segments_of(result), @@ -410,7 +456,11 @@ def result_of(result: Any) -> dict[str, Any]: # Platform evaluations are backend-owned. Harness checks are direct # execution evidence and stay namespaced in metadata rather than # being sent through the removed SDK `evaluations` input field. - "harness_evaluations": evaluations_of(result), + "harness_evaluations": evaluations, + "harness_eval_coverage": evaluation_coverage, + "harness_failure_classification": ( + "grading_failure" if grading_failures else "" + ), "harness_spent_usd": round( float(getattr(result, "spent_usd", 0.0) or 0.0), 4 ), diff --git a/src/fi/alk/harness/provision.py b/src/fi/alk/harness/provision.py index e326105b..cc716a0b 100644 --- a/src/fi/alk/harness/provision.py +++ b/src/fi/alk/harness/provision.py @@ -37,6 +37,11 @@ can_generate_runtime, prepare_generated_runtime, ) +from .environment_resolution import ( + ResolvedEnvironmentPlan, + resolve_environment_plan, + write_environment_resolution, +) from .service_catalog import address, profile_for from .secrets import runtime_configuration_value @@ -81,6 +86,19 @@ class ProvisionError(RuntimeError): """The source environment could not be discovered, started, or inspected.""" + def __init__( + self, + message: str, + *, + code: str = "environment_provision_failed", + action: str = "", + details: dict[str, str] | None = None, + ) -> None: + super().__init__(message) + self.code = code + self.action = action + self.details = details or {} + @dataclass class ProvisionedEnvironment: @@ -111,6 +129,11 @@ class ProvisionedEnvironment: # Names only. Values are resolved from the job secret environment immediately before the # ephemeral worker starts and are never serialized into environment.json or a bundle. runtime_configuration_names: list[str] = field(default_factory=list) + # Stable admission decision. These values are safe to persist and make failures/replays + # explainable without relying on private Docker logs or re-inferring repository packaging. + packaging_type: str = "unknown" + runtime_adapter: str = "unknown" + selected_runtime: str = "" def save(self, destination: Path) -> Path: destination.mkdir(parents=True, exist_ok=True) @@ -139,6 +162,19 @@ def compose_file(source: str | Path) -> Path | None: return None +def environment_adapter_context( + source: str | Path, contract: Any | None +) -> dict[str, str]: + """Return the stable packaging/adapter decision used for audit and customer errors.""" + plan = resolve_environment_plan(source, contract=contract) + return { + "packaging_type": plan.packaging_type, + "runtime_adapter": plan.runtime_adapter, + "failed_adapter": plan.runtime_adapter, + "selected_runtime": plan.selected_runtime, + } + + _FINGERPRINT_IGNORED = { ".git", ".mypy_cache", @@ -546,10 +582,9 @@ def _write_port_override( # Compose excludes profiled services unless that profile is explicitly enabled. Do not # allocate ports for dormant TURN/admin/debug services; doing so can exhaust a runner's # ephemeral port range before the selected environment even starts. - if not _service_starts_by_default(service): - continue + starts_by_default = _service_starts_by_default(service) ports: list[dict[str, Any]] = [] - for item in service.get("ports") or []: + for item in service.get("ports") or [] if starts_by_default else []: target = int(item.get("target") or 0) if not target or target in dynamic_targets: continue @@ -565,20 +600,39 @@ def _write_port_override( } ) fixed_container_name = bool(str(service.get("container_name") or "").strip()) - if ports or fixed_container_name: + # Label default services and submitted build runtimes. Dormant third-party profiles are + # not part of this admitted environment and must remain untouched. + if ports or fixed_container_name or starts_by_default or service.get("build"): services.append((name, ports, fixed_container_name)) reset_env_files = _missing_env_file_services(Path(environment.compose_file)) if not services and not reset_env_files: return + service_overrides = { + name: (ports, fixed_container_name) + for name, ports, fixed_container_name in services + } + ordered_names = list( + dict.fromkeys([*reset_env_files, *(name for name, *_ in services)]) + ) lines = ["services:"] - for name in reset_env_files: - lines.extend((f" {json.dumps(name)}:", " env_file: !reset []")) - for name, ports, fixed_container_name in services: + for name in ordered_names: lines.append(f" {json.dumps(name)}:") + if name in reset_env_files: + lines.append(" env_file: !reset []") + service_override = service_overrides.get(name) + if service_override is None: + continue + ports, fixed_container_name = service_override if fixed_container_name: # A submitted container_name bypasses Compose project scoping and collides across # concurrent jobs. Compose's reset tag restores its normal project-owned name. lines.append(" container_name: !reset null") + lines.extend( + ( + " labels:", + f" com.futureagi.harness.project: {json.dumps(environment.project)}", + ) + ) if ports: lines.append(" ports: !override") for item in ports: @@ -1773,6 +1827,31 @@ def provision( source_root = Path(source).expanduser().resolve() destination = Path(destination) packaging = inspect_packaging(source_root) + fingerprint = source_fingerprint(source_root) + resolved_plan: ResolvedEnvironmentPlan = resolve_environment_plan( + source_root, + packaging, + contract, + source_fingerprint=fingerprint, + ) + write_environment_resolution(destination, resolved_plan) + if not resolved_plan.supported: + raise ProvisionError( + resolved_plan.message or "environment plan is unsupported", + code=resolved_plan.code or "environment_plan_unsupported", + action=resolved_plan.action, + details={ + "packaging_type": resolved_plan.packaging_type, + "failed_adapter": resolved_plan.runtime_adapter, + "selected_runtime": resolved_plan.selected_runtime, + "environment_plan_hash": resolved_plan.digest, + }, + ) + adapter_context = { + "packaging_type": resolved_plan.packaging_type, + "runtime_adapter": resolved_plan.runtime_adapter, + "selected_runtime": resolved_plan.selected_runtime, + } generated_runtime: GeneratedRuntimePlan | None = None explicit_compose = str( getattr(getattr(contract, "runtime", None), "compose_file", "") or "" @@ -1901,7 +1980,12 @@ def provision( raise ProvisionError( f"{source_root} does not ship a Compose file; a non-Compose runtime adapter is required" ) - fingerprint = source_fingerprint(source_root) + if source_fingerprint(source_root) != resolved_plan.source_fingerprint: + raise ProvisionError( + "submitted source changed after environment planning", + code="environment_plan_source_fingerprint_mismatch", + action="Retry from a fresh immutable source snapshot.", + ) runtime_fingerprint = generated_runtime.fingerprint if generated_runtime else "" runtime_configuration_names = sorted( { @@ -1930,6 +2014,9 @@ def provision( existing.overrides = _overrides(existing, config) existing.internal_overrides = _internal_overrides(existing, config) existing.runtime_configuration_names = runtime_configuration_names + existing.packaging_type = adapter_context["packaging_type"] + existing.runtime_adapter = adapter_context["runtime_adapter"] + existing.selected_runtime = adapter_context["selected_runtime"] existing.save(destination) return existing @@ -1960,6 +2047,9 @@ def provision( ), runtime_fingerprint=runtime_fingerprint, runtime_configuration_names=runtime_configuration_names, + packaging_type=adapter_context["packaging_type"], + runtime_adapter=adapter_context["runtime_adapter"], + selected_runtime=adapter_context["selected_runtime"], ) _write_initial_env_file_override(destination, environment) config = _config(environment) diff --git a/src/fi/alk/harness/run/grade.py b/src/fi/alk/harness/run/grade.py index d200ff99..581439cd 100644 --- a/src/fi/alk/harness/run/grade.py +++ b/src/fi/alk/harness/run/grade.py @@ -18,6 +18,7 @@ import json import logging +import os from dataclasses import dataclass, field from typing import Any @@ -57,6 +58,7 @@ class Checkpoint: detail: str = "" # The eval that decided it, where one did. Empty for anything settled by code or judged here. by: str = "" + grading_error: bool = False def line(self) -> str: return f" [{'x' if self.passed else ' '}] {self.kind}: {self.name}" + ( @@ -72,6 +74,10 @@ class Judgement: why: str = "" # Which eval decided this, when it was decided by one rather than here. by: str = "" + # True means the grading system failed to produce a verdict. This is an + # infrastructure/product failure, not evidence that the submitted agent + # failed the claim. + grading_error: bool = False @dataclass @@ -116,6 +122,10 @@ class Result: def conduct_failures(self) -> list[Judgement]: return [item for item in self.conduct if not item.holds] + @property + def grading_failures(self) -> list[Judgement]: + return [item for item in self.conduct if item.grading_error] + @property def passed(self) -> bool: return ( @@ -276,12 +286,22 @@ def judge_suite_evals( """ from . import platform_evals - if ( - contract.modality != "voice" - or not suite_evals - or not platform_evals.configured() - ): + if contract.modality != "voice" or not suite_evals: return [] + hosted = os.getenv("ALK_HOSTED_EXECUTION", "") == "1" + if not platform_evals.configured(): + if not hosted: + return [] + return [ + Judgement( + claim=suite_eval.name, + kind=suite_eval.name, + holds=False, + why="Required platform evaluation is not configured for this hosted job.", + grading_error=True, + ) + for suite_eval in suite_evals + ] verdicts: list[Judgement] = [] for suite_eval in suite_evals: inputs = { @@ -295,6 +315,17 @@ def judge_suite_evals( suite_eval.name, ", ".join(missing), ) + if hosted: + verdicts.append( + Judgement( + claim=suite_eval.name, + kind=suite_eval.name, + holds=False, + why="Required evaluation inputs were unavailable: " + + ", ".join(missing), + grading_error=True, + ) + ) continue try: answered = platform_evals.judge_builtin( @@ -305,6 +336,16 @@ def judge_suite_evals( logging.getLogger(__name__).warning( "platform suite eval %s unavailable: %s", suite_eval.name, failed ) + if hosted: + verdicts.append( + Judgement( + claim=suite_eval.name, + kind=suite_eval.name, + holds=False, + why=f"Required platform evaluation could not run: {failed}", + grading_error=True, + ) + ) continue output = answered["output"] choice = output.get("choice") if isinstance(output, dict) else None @@ -487,6 +528,7 @@ def checkpoints(settled: list[Outcome], judged: list[Judgement]) -> list[Checkpo passed=item.holds, detail=item.why, by=item.by, + grading_error=item.grading_error, ) for item in judged ) diff --git a/src/fi/alk/harness/sandbox_server.py b/src/fi/alk/harness/sandbox_server.py index cc31c748..57fc2d90 100644 --- a/src/fi/alk/harness/sandbox_server.py +++ b/src/fi/alk/harness/sandbox_server.py @@ -43,7 +43,13 @@ discover_credentials, ) from .executor import GitHubSourceAcquirer, SourceAcquisitionError +from .failure_reporting import ( + FAILURE_PATH_ENVIRONMENT, + load_stage_failure, + sanitize_failure_message, +) from .github import parse_github_location +from .environment_resolution import ResolvedEnvironmentPlan, resolve_environment_plan from .job import ( AgentConnection, ExecutionMode, @@ -78,6 +84,7 @@ class LocalSandboxRequest(BaseModel): connector_config: dict[str, Any] = Field(default_factory=dict) secret_refs: dict[str, SecretRef] = Field(default_factory=dict) environment_values: dict[str, SecretStr] = Field(default_factory=dict) + controller_environment_values: dict[str, SecretStr] = Field(default_factory=dict) platform_run_id: str | None = None metadata: dict[str, Any] = Field(default_factory=dict) @@ -92,6 +99,7 @@ def _one_source(self) -> "LocalSandboxRequest": ): raise ValueError("exactly_one_source_required") _validate_environment_values(self.environment_values, self.secret_refs) + _validate_controller_environment_values(self.controller_environment_values) return self @@ -129,11 +137,13 @@ class SandboxRerunRequest(BaseModel): secret_refs: dict[str, SecretRef] = Field(default_factory=dict) environment_values: dict[str, SecretStr] = Field(default_factory=dict) + controller_environment_values: dict[str, SecretStr] = Field(default_factory=dict) only: list[str] = Field(default_factory=list) @model_validator(mode="after") def _valid_environment(self) -> "SandboxRerunRequest": _validate_environment_values(self.environment_values, self.secret_refs) + _validate_controller_environment_values(self.controller_environment_values) return self @@ -144,6 +154,7 @@ class SandboxPreflightResponse(BaseModel): checkout_required: bool = False credentials: CredentialManifest packaging: PackagingManifest | None = None + environment_plan: ResolvedEnvironmentPlan | None = None notes: list[str] = Field(default_factory=list) @@ -227,6 +238,7 @@ def __init__( # Values supplied through the platform's .env flow live only for the lifetime of the # job. Persisted job/state artifacts contain the opaque mounted references created below. self._ephemeral_secrets: dict[str, dict[str, str]] = {} + self._ephemeral_controller_secrets: dict[str, dict[str, str]] = {} self._ephemeral_secret_file_names: dict[str, set[str]] = {} self._recover_orphans() @@ -477,6 +489,19 @@ def submit(self, request: LocalSandboxRequest) -> SandboxJobResponse: purpose=f"job-scoped environment value for {name}", ) mounted_values[internal_key] = value.get_secret_value() + controller_refs: dict[str, SecretRef] = {} + for index, (name, value) in enumerate( + request.controller_environment_values.items() + ): + internal_key = ( + f"ALK_CONTROLLER_{identifier.replace('-', '').upper()}_{index}" + ) + controller_refs[name] = SecretRef( + manager="mounted", + key=internal_key, + purpose=f"job-scoped harness controller value for {name}", + ) + mounted_values[internal_key] = value.get_secret_value() claimed_files, secret_file_names = self._claim_secret_files( identifier, request.secret_refs ) @@ -514,7 +539,11 @@ def submit(self, request: LocalSandboxRequest) -> SandboxJobResponse: agent=AgentConnection( connector=request.connector, config=request.connector_config, - secret_refs={**request.secret_refs, **mounted_refs}, + secret_refs={ + **request.secret_refs, + **mounted_refs, + **controller_refs, + }, ), scenario_count=request.scenario_count, seed=request.seed, @@ -598,6 +627,10 @@ def rerun(self, job_id: str, request: SandboxRerunRequest) -> SandboxJobResponse name: value.get_secret_value() for name, value in request.environment_values.items() } + controller_configuration = { + name: value.get_secret_value() + for name, value in request.controller_environment_values.items() + } claimed_files, secret_file_names = self._claim_secret_files( job_id, request.secret_refs ) @@ -610,7 +643,8 @@ def rerun(self, job_id: str, request: SandboxRerunRequest) -> SandboxJobResponse self._delete_job_secret_files(job_id) raise runtime_configuration.update(resolved) - self._ephemeral_secrets[job_id] = dict(runtime_configuration) + self._ephemeral_secrets[job_id] = runtime_configuration + self._ephemeral_controller_secrets[job_id] = controller_configuration self._ephemeral_secret_file_names[job_id] = secret_file_names state_path = self.jobs_root / job_id / "state.json" @@ -657,13 +691,18 @@ async def _execute_rerun(self, job: HarnessJob, only: list[str]) -> None: ) _write_json(state_path, state) log_handle = (directory / "worker.log").open("ab") + failure_path = directory / "stage-failure.json" try: shutil.rmtree(replay_root, ignore_errors=True) replay_root.parent.mkdir(parents=True, exist_ok=True) shutil.copytree(output / "environment-bundle", replay_root) runtime_configuration = self._ephemeral_secrets.get(job.job_id, {}) + controller_configuration = self._ephemeral_controller_secrets.get( + job.job_id, {} + ) child_environment = worker_environment( - {}, runtime_configuration=runtime_configuration + controller_configuration, + runtime_configuration=runtime_configuration, ) child_environment["ALK_RUNTIME_CONFIGURATION_NAMES"] = ",".join( sorted(runtime_configuration) @@ -676,16 +715,23 @@ async def _execute_rerun(self, job: HarnessJob, only: list[str]) -> None: ) child_environment["ALK_HARNESS_JOB_ID"] = job.job_id - async def run_stage(command: list[str]) -> int: + async def run_stage( + command: list[str], + ) -> tuple[int, dict[str, Any] | None]: + failure_path.unlink(missing_ok=True) + stage_environment = { + **child_environment, + FAILURE_PATH_ENVIRONMENT: str(failure_path), + } process = await asyncio.create_subprocess_exec( *command, stdout=log_handle, stderr=asyncio.subprocess.STDOUT, start_new_session=True, - env=child_environment, + env=stage_environment, ) self._processes[job.job_id] = process - return await process.wait() + return await process.wait(), load_stage_failure(failure_path) environment_up = [ sys.executable, @@ -720,14 +766,20 @@ async def run_stage(command: list[str]) -> int: str(output), ] - up_code = await run_stage(environment_up) + up_code, up_failure = await run_stage(environment_up) simulation_code = 1 + simulation_failure = None cleanup_code = 0 + cleanup_failure = None if up_code == 0: try: - simulation_code = await run_stage(simulation) + simulation_code, simulation_failure = await run_stage( + simulation + ) finally: - cleanup_code = await run_stage(environment_down) + cleanup_code, cleanup_failure = await run_stage( + environment_down + ) # Exit 2 is a completed suite whose submitted agent failed checks. successful_execution = ( @@ -754,6 +806,13 @@ async def run_stage(command: list[str]) -> int: if cleanup_code != 0 else simulation_code ) + structured_failure = ( + up_failure + if up_code != 0 + else cleanup_failure + if cleanup_code != 0 + else simulation_failure + ) or {} state = _read_json(state_path) if state.get("stage") != HarnessStage.CANCELED.value: state.update( @@ -775,23 +834,32 @@ async def run_stage(command: list[str]) -> int: else { "domain": "environment", "stage": failure_stage, - "code": "saved_session_rerun_failed", - "message": f"{failed_stage} exited {return_code}", - "retryable": False, + "code": structured_failure.get( + "code", "saved_session_rerun_failed" + ), + "message": structured_failure.get( + "detail", f"{failed_stage} exited {return_code}" + ), + "retryable": bool( + structured_failure.get("retryable", False) + ), + "details": structured_failure.get("details", {}), + "action": structured_failure.get("action", ""), }, updated_at=_now(), ) _write_json(state_path, state) except Exception as exc: + safe_message = sanitize_failure_message(exc) state = _read_json(state_path) state.update( stage=HarnessStage.FAILED.value, - detail=f"{type(exc).__name__}: {exc}", + detail=f"{type(exc).__name__}: {safe_message}", failure={ "domain": "infrastructure", "stage": HarnessStage.RUNNING.value, "code": type(exc).__name__, - "message": str(exc), + "message": safe_message, "retryable": False, }, updated_at=_now(), @@ -802,8 +870,10 @@ async def run_stage(command: list[str]) -> int: self._processes.pop(job.job_id, None) self._tasks.pop(job.job_id, None) self._ephemeral_secrets.pop(job.job_id, None) + self._ephemeral_controller_secrets.pop(job.job_id, None) self._ephemeral_secret_file_names.pop(job.job_id, None) self._delete_job_secret_files(job.job_id) + failure_path.unlink(missing_ok=True) shutil.rmtree(replay_root, ignore_errors=True) def preflight(self, request: SandboxPreflightRequest) -> SandboxPreflightResponse: @@ -829,6 +899,11 @@ def preflight(self, request: SandboxPreflightRequest) -> SandboxPreflightRespons }, scan_paths=_credential_scan_paths(packaging), ) + environment_plan = resolve_environment_plan( + source, + packaging, + source_fingerprint=manifest.source_digest, + ) return SandboxPreflightResponse( source_kind=( SourceKind.ARCHIVE @@ -840,11 +915,11 @@ def preflight(self, request: SandboxPreflightRequest) -> SandboxPreflightRespons ), ready_to_submit=( manifest.ready - and packaging.ready - and (packaging.agent_runtime_packaged or not packaging.candidates) + and environment_plan.execution_ready ), credentials=manifest, packaging=packaging, + environment_plan=environment_plan, ) location = _github_location(request.github_repository or "") repository = location.repository @@ -1127,6 +1202,7 @@ async def _execute(self, job: HarnessJob, source: Path | None) -> None: ) _write_json(state_path, state) except Exception as exc: + safe_message = sanitize_failure_message(exc, source=source) state = _read_json(state_path) domain = ( "connectivity" @@ -1135,12 +1211,12 @@ async def _execute(self, job: HarnessJob, source: Path | None) -> None: ) state.update( stage=HarnessStage.FAILED.value, - detail=f"{type(exc).__name__}: {exc}", + detail=f"{type(exc).__name__}: {safe_message}", failure={ "domain": domain, "stage": HarnessStage.ACQUIRING_SOURCE.value, "code": type(exc).__name__, - "message": str(exc), + "message": safe_message, "retryable": isinstance(exc, SourceAcquisitionError), }, updated_at=_now(), @@ -1151,6 +1227,7 @@ async def _execute(self, job: HarnessJob, source: Path | None) -> None: self._processes.pop(job.job_id, None) self._tasks.pop(job.job_id, None) self._ephemeral_secrets.pop(job.job_id, None) + self._ephemeral_controller_secrets.pop(job.job_id, None) self._ephemeral_secret_file_names.pop(job.job_id, None) self._delete_job_secret_files(job.job_id) @@ -1178,7 +1255,16 @@ def get(self, job_id: str) -> SandboxJobResponse: # source of truth, so expose its timestamp and stage instead of making # a healthy long-running job look stale in the platform UI. updated_at = events[-1].get("wall_time") or updated_at - detail = { + event_detail = next( + ( + str(event.get("payload", {}).get("detail")) + for event in reversed(events) + if event.get("payload", {}).get("detail") + and _stage_from_events([event]) == event_stage + ), + None, + ) + detail = event_detail or { HarnessStage.UNDERSTANDING_AGENT.value: "understanding agent source", HarnessStage.GENERATING_ENVIRONMENT.value: "provisioning and validating environment", HarnessStage.GENERATING_SCENARIOS.value: "generating and validating scenarios", @@ -1321,6 +1407,7 @@ async def cancel(self, job_id: str) -> SandboxJobResponse: ) _write_json(state_path, state) self._ephemeral_secrets.pop(job_id, None) + self._ephemeral_controller_secrets.pop(job_id, None) self._ephemeral_secret_file_names.pop(job_id, None) self._delete_job_secret_files(job_id) return self.get(job_id) @@ -1386,6 +1473,30 @@ def _configuration_environment(config: dict[str, Any]) -> dict[str, str]: "PATH", "PYTHONPATH", } +_CONTROLLER_ENVIRONMENT_NAMES = { + "FI_API_KEY", + "FI_SECRET_KEY", + "HARNESS_PLATFORM_API_KEY", + "HARNESS_PLATFORM_SECRET_KEY", + "HARNESS_PLATFORM_WORKSPACE_ID", +} + + +def _validate_controller_environment_values(values: dict[str, SecretStr]) -> None: + """Accept only reporting identity supplied by the trusted control plane. + + This field is not part of the customer source/environment contract. Keeping + a strict allow-list prevents the platform-to-provider seam from becoming a + general way to mutate the harness controller process. + """ + + unsupported = sorted(set(values) - _CONTROLLER_ENVIRONMENT_NAMES) + if unsupported: + raise ValueError( + "controller_environment_name_unsupported: " + ", ".join(unsupported) + ) + if sum(len(name) + len(value.get_secret_value()) for name, value in values.items()) > 4096: + raise ValueError("controller_environment_values_size_exceeded") def _validate_environment_values( @@ -1677,13 +1788,28 @@ def _adjustments(directory: Path) -> list[dict[str, Any]]: def _scenario_delta(instruction: str) -> int | None: match = re.search( - r"\b(?:add|create|generate|write)\s+(\d{1,3})\s+(?:more\s+)?scenarios?\b", + r"\b(?:add|create|generate|write)\s+" + r"(\d{1,3}|one|two|three|four|five|six|seven|eight|nine|ten)\s+" + r"(?:more\s+)?scenarios?\b", instruction, re.IGNORECASE, ) if not match: return None - return min(100, int(match.group(1))) + raw_count = match.group(1).lower() + word_counts = { + "one": 1, + "two": 2, + "three": 3, + "four": 4, + "five": 5, + "six": 6, + "seven": 7, + "eight": 8, + "nine": 9, + "ten": 10, + } + return min(100, word_counts.get(raw_count, int(raw_count) if raw_count.isdigit() else 0)) def _adjustment_stage(instruction: str, current_stage: str) -> str: diff --git a/tests/test_harness.py b/tests/test_harness.py index 608237f7..28b0f550 100644 --- a/tests/test_harness.py +++ b/tests/test_harness.py @@ -261,9 +261,10 @@ def run(_environment, *arguments, **_kwargs): assert {"LIVEKIT_API_KEY", "LIVEKIT_URL", "DEEPGRAM_API_KEY"} <= set( environment.runtime_configuration_names ) - assert (output / "compose.harness.override.yaml").read_text() == ( - 'services:\n "agent":\n env_file: !reset []\n' - ) + override = (output / "compose.harness.override.yaml").read_text() + assert override.count(' "agent":') == 1 + assert "env_file: !reset []" in override + assert "com.futureagi.harness.project:" in override def test_source_environment_uses_configured_docker_gateway(tmp_path, monkeypatch): @@ -6365,6 +6366,30 @@ def judge_builtin(name, inputs): assert all(verdict.holds for verdict in verdicts) +def test_hosted_voice_marks_unavailable_required_evals_as_grading_failures( + monkeypatch, +): + from fi.alk.harness.catalogue import default_suite_evals + from fi.alk.harness.contract import AgentContract + from fi.alk.harness.run import platform_evals + from fi.alk.harness.run.conversation import Transcript + from fi.alk.harness.run.grade import judge_suite_evals + from fi.alk.harness.scenario import Scenario + + monkeypatch.setenv("ALK_HOSTED_EXECUTION", "1") + monkeypatch.setattr(platform_evals, "configured", lambda: False) + verdicts = judge_suite_evals( + default_suite_evals(), + Scenario(name="support", instruction="help", sub_goals=[]), + Transcript(), + AgentContract(agent="voice", modality="voice"), + ) + + assert len(verdicts) == len(default_suite_evals()) + assert all(verdict.grading_error for verdict in verdicts) + assert all(not verdict.holds for verdict in verdicts) + + def test_suite_evals_do_not_run_for_non_voice_agents(monkeypatch): from fi.alk.harness.catalogue import default_suite_evals from fi.alk.harness.contract import AgentContract @@ -6613,6 +6638,71 @@ def test_a_scenario_that_never_ran_is_not_reported_as_one_the_agent_failed(): assert "not ready" in blocked["error_message"] +def test_a_grading_failure_does_not_rewrite_completed_call_status(): + from fi.alk.harness import platform + from fi.alk.harness.run.grade import Checkpoint, Judgement + + result = _reported_result( + checkpoints=[ + Checkpoint( + name="task completion", + kind="eval", + passed=False, + detail="evaluator unavailable", + grading_error=True, + ) + ], + conduct=[ + Judgement( + claim="task completion", + kind="eval", + holds=False, + why="evaluator unavailable", + grading_error=True, + ) + ] + ) + + payload = platform.result_of(result) + + assert payload["status"] == "completed" + assert "error_message" not in payload + assert payload["call_metadata"]["harness_eval_coverage"] == { + "expected": 1, + "executed": 0, + "failed": 1, + "complete": False, + } + + +def test_platform_call_duration_excludes_connection_and_retry_waits(): + from fi.alk.harness import platform + + connected = _reported_result(seconds=255.0) + connected.exchanges = [ + { + "speaker": "agent", + "text": "hello", + "start_time_ms": 1_000, + "end_time_ms": 4_000, + }, + { + "speaker": "customer", + "text": "goodbye", + "start_time_ms": 102_000, + "end_time_ms": 107_205, + }, + ] + unavailable = _reported_result( + seconds=301.1, + exchanges=[], + problems=["the target worker never joined"], + ) + + assert platform.result_of(connected)["duration_seconds"] == 106 + assert platform.result_of(unavailable)["duration_seconds"] == 0 + + def test_a_second_run_joins_the_same_test_rather_than_starting_another(): from fi.alk.harness import platform diff --git a/tests/test_harness_architecture.py b/tests/test_harness_architecture.py index 7540d0df..5de5c718 100644 --- a/tests/test_harness_architecture.py +++ b/tests/test_harness_architecture.py @@ -27,8 +27,13 @@ HarnessExecutor, _failure_from_events, ) +from fi.alk.harness.failure_reporting import ( + clear_stage_failure, + load_stage_failure, + record_stage_failure, +) from fi.alk.harness.job import FailureDomain, HarnessFailure, HarnessJob, HarnessStage -from fi.alk.harness.provision import source_fingerprint +from fi.alk.harness.provision import environment_adapter_context, source_fingerprint from fi.simulate.runtime.spec import RuntimeIsolation from fi.simulate.runtime.events import CanonicalEvent @@ -159,6 +164,57 @@ async def failed_understanding(_args) -> int: assert cleaned == [output.resolve()] +def test_autonomous_pipeline_emits_safe_progress_and_failure_detail( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + from fi.alk.harness import cli + from fi.alk.harness.failure_reporting import record_stage_failure + + source = tmp_path / "agent" + source.mkdir() + (source / "agent.py").write_text("print('agent')\n", encoding="utf-8") + output = tmp_path / "artifacts" + + async def failed_understanding(_args) -> int: + record_stage_failure( + "understanding_contract_missing", + "The submitted source did not produce an agent contract", + ) + return 1 + + monkeypatch.setattr(cli, "_understand", failed_understanding) + monkeypatch.setattr("fi.alk.harness.provision.stop", lambda _destination: False) + + status = asyncio.run( + cli._auto( + SimpleNamespace( + path=str(source), + name="agent", + kind="repo", + out=str(output), + count=1, + model=None, + run_model=None, + ) + ) + ) + + events = [ + json.loads(line) + for line in (output / "harness-events.jsonl") + .read_text(encoding="utf-8") + .splitlines() + ] + progress = next( + event for event in events if event["type"] == "harness.stage.progress" + ) + failed = next(event for event in events if event["type"] == "harness.stage.failed") + assert status == 1 + assert progress["payload"]["detail"].startswith("Understanding source · inspecting") + assert failed["payload"]["code"] == "understanding_contract_missing" + assert "agent contract" in failed["payload"]["detail"] + + def test_bundle_is_content_addressed_and_reproducible(tmp_path: Path) -> None: (tmp_path / "compose.yaml").write_text("services: {}\n", encoding="utf-8") first = seal_bundle(tmp_path, _manifest()) @@ -169,6 +225,29 @@ def test_bundle_is_content_addressed_and_reproducible(tmp_path: Path) -> None: assert first.files[0].path == "compose.yaml" +def test_environment_adapter_decision_is_stable_for_customer_errors_and_plans( + tmp_path: Path, +) -> None: + (tmp_path / "Dockerfile").write_text("FROM python:3.12-slim\n", encoding="utf-8") + contract = SimpleNamespace( + runtime=SimpleNamespace(compose_file="", dockerfile="Dockerfile") + ) + + first = environment_adapter_context(tmp_path, contract) + second = environment_adapter_context(tmp_path, contract) + + assert ( + first + == second + == { + "packaging_type": "dockerfile", + "runtime_adapter": "managed_compose_for_dockerfile", + "failed_adapter": "managed_compose_for_dockerfile", + "selected_runtime": "Dockerfile", + } + ) + + def test_bundle_detects_file_tampering(tmp_path: Path) -> None: compose = tmp_path / "compose.yaml" compose.write_text("services: {}\n", encoding="utf-8") @@ -329,7 +408,9 @@ def test_environment_up_rejects_a_corrupt_bundle_before_provisioning( ) monkeypatch.setattr( "fi.alk.harness.provision.provision", - lambda *_args, **_kwargs: pytest.fail("corrupt bundle reached Docker provisioning"), + lambda *_args, **_kwargs: pytest.fail( + "corrupt bundle reached Docker provisioning" + ), ) result = asyncio.run( @@ -517,6 +598,57 @@ def test_failed_stage_is_reported_as_structured_non_retryable_failure(tmp_path: assert failure.retryable is False +def test_failed_stage_preserves_safe_packaging_and_adapter_details(tmp_path: Path): + (tmp_path / "harness-events.jsonl").write_text( + '{"type":"harness.stage.failed","payload":' + '{"stage":"environment","status":1,' + '"code":"environment_provision_failed",' + '"detail":"Cannot create the source environment: no runtime",' + '"details":{"packaging_type":"unpackaged",' + '"failed_adapter":"generated_runtime"}}}\n', + encoding="utf-8", + ) + + failure = _failure_from_events(tmp_path) + + assert failure.message.endswith("no runtime") + assert failure.details == { + "packaging_type": "unpackaged", + "failed_adapter": "generated_runtime", + "status": 1, + } + + +def test_stage_failure_sidecar_is_sanitized_and_structured( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + target = tmp_path / "failure.json" + source = tmp_path / "private-checkout" + source.mkdir() + monkeypatch.setenv("ALK_HARNESS_FAILURE_PATH", str(target)) + monkeypatch.setenv("VENDOR_API_KEY", "super-secret-value") + clear_stage_failure() + + record_stage_failure( + "environment_provision_failed", + f"{source} rejected api_key=super-secret-value", + source=source, + action="Provide a supported runtime without api_key=super-secret-value.", + details={ + "packaging_type": "dockerfile", + "failed_adapter": "managed_compose_for_dockerfile", + }, + ) + + failure = load_stage_failure(target) + assert failure is not None + assert str(source) not in failure["detail"] + assert "super-secret-value" not in failure["detail"] + assert "super-secret-value" not in failure["action"] + assert failure["action"] == "Provide a supported runtime without api_key=[REDACTED]" + assert failure["details"]["packaging_type"] == "dockerfile" + + def test_only_agent_failures_are_owned_by_the_customer() -> None: agent = HarnessFailure( domain=FailureDomain.AGENT, diff --git a/tests/test_harness_environment_resolution.py b/tests/test_harness_environment_resolution.py new file mode 100644 index 00000000..ec369fb5 --- /dev/null +++ b/tests/test_harness_environment_resolution.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from fi.alk.harness.environment_resolution import resolve_environment_plan +from fi.alk.harness.packaging import inspect_packaging + + +def _runtime(**values): + defaults = { + "compose_file": "", + "dockerfile": "", + "workdir": "", + "language": "", + "version": "", + "command": [], + } + return SimpleNamespace(**{**defaults, **values}) + + +@pytest.mark.parametrize( + ("case", "expected_type", "expected_adapter"), + [ + ("uber-compose", "compose", "submitted_compose"), + ("hotel-dockerfile", "dockerfile", "managed_compose_for_dockerfile"), + ("livekit-dockerfile", "dockerfile", "managed_compose_for_dockerfile"), + ("packaged-chat", "dockerfile", "managed_compose_for_dockerfile"), + ("unpackaged-chat", "unpackaged", "generated_runtime"), + ("complex-compose", "compose", "submitted_compose"), + ], +) +def test_environment_certification_static_matrix( + tmp_path: Path, case: str, expected_type: str, expected_adapter: str +) -> None: + source = tmp_path / case + source.mkdir() + runtime = _runtime() + dependencies = [] + if case in {"uber-compose", "complex-compose"}: + (source / "Dockerfile").write_text("FROM python:3.12-slim\n", encoding="utf-8") + services = " agent:\n build: .\n command: [python, agent.py]\n" + if case == "complex-compose": + services += ( + " postgres:\n image: postgres:17\n" + " redis:\n image: redis:7\n" + " minio:\n image: minio/minio:latest\n" + ) + (source / "compose.yml").write_text("services:\n" + services, encoding="utf-8") + runtime = _runtime(compose_file="compose.yml") + elif case == "unpackaged-chat": + (source / "requirements.txt").write_text("fastapi==0.115.0\n", encoding="utf-8") + (source / "main.py").write_text("print('ready')\n", encoding="utf-8") + runtime = _runtime(language="python", command=["python", "main.py"]) + else: + (source / "Dockerfile").write_text("FROM python:3.12-slim\n", encoding="utf-8") + runtime = _runtime(dockerfile="Dockerfile") + if case == "hotel-dockerfile": + dependencies = [ + SimpleNamespace( + name="local SQLite state", + engine="sqlite", + kind="datastore", + what="in-process booking state", + reached=SimpleNamespace( + loader_module="hotel.db", loader_function="load" + ), + ) + ] + contract = SimpleNamespace( + runtime=runtime, dependencies=dependencies, data_store=None + ) + packaging = inspect_packaging(source) + + first = resolve_environment_plan( + source, packaging, contract, source_fingerprint="a" * 64 + ) + second = resolve_environment_plan( + source, packaging, contract, source_fingerprint="a" * 64 + ) + + assert first.supported is True + assert first.packaging_type == expected_type + assert first.runtime_adapter == expected_adapter + assert first.digest == second.digest + assert first.source_fingerprint == "a" * 64 + + +def test_environment_resolution_rejects_unowned_dependency_before_docker( + tmp_path: Path, +) -> None: + (tmp_path / "requirements.txt").write_text("fastapi==0.115.0\n", encoding="utf-8") + (tmp_path / "main.py").write_text("print('ready')\n", encoding="utf-8") + contract = SimpleNamespace( + runtime=_runtime(language="python", command=["python", "main.py"]), + data_store=None, + dependencies=[ + SimpleNamespace( + name="private mystery service", + engine="mysteryd", + kind="service", + what="required remote state", + reached=SimpleNamespace(), + ) + ], + ) + + plan = resolve_environment_plan(tmp_path, contract=contract) + + assert plan.supported is False + assert plan.code == "unsupported_dependency" + assert plan.dependencies[0].ownership == "unsupported" + assert plan.action + + +def test_environment_resolution_accepts_optional_external_api_with_embedded_fallback( + tmp_path: Path, +) -> None: + (tmp_path / "pyproject.toml").write_text( + '[project]\nname = "frontdesk"\nversion = "0.1.0"\ndependencies = ["livekit-agents"]\n', + encoding="utf-8", + ) + (tmp_path / "agent.py").write_text("print('ready')\n", encoding="utf-8") + contract = SimpleNamespace( + runtime=_runtime(language="python", command=["python", "agent.py"]), + data_store=None, + dependencies=[ + SimpleNamespace( + name="FakeCalendar", + engine="in-process Python", + kind="datastore", + what="fallback used when CAL_API_KEY is unset", + reached=SimpleNamespace( + loader_module="calendar_api", + loader_function="FakeCalendar", + dsn_env="", + config_key="", + password_from="", + database="", + ), + ), + SimpleNamespace( + name="cal.com API", + engine="HTTP REST", + kind="service", + what="optional production calendar backend", + reached=SimpleNamespace( + loader_module="", + loader_function="", + dsn_env="CAL_API_KEY", + config_key="", + password_from="", + database="", + ), + ), + ], + ) + + plan = resolve_environment_plan(tmp_path, contract=contract) + + assert plan.supported is True + assert plan.runtime_adapter == "generated_runtime" + assert [item.ownership for item in plan.dependencies] == [ + "embedded", + "external_provider", + ] diff --git a/tests/test_harness_sandbox_server.py b/tests/test_harness_sandbox_server.py index e988c322..a1cb944c 100644 --- a/tests/test_harness_sandbox_server.py +++ b/tests/test_harness_sandbox_server.py @@ -15,12 +15,27 @@ _CONTROLLER_TOKEN, _presentation_value, _process_identity, + _scenario_delta, _worker_failure_retryable, create_app, ) from fi.alk.harness.secrets import resolve_worker_secrets +@pytest.mark.parametrize( + ("instruction", "expected"), + [ + ("Add 10 more scenarios covering payment failures", 10), + ( + "generate one more scenario where the user asks for a discount", + 1, + ), + ], +) +def test_scenario_delta_accepts_numeric_and_word_counts(instruction, expected): + assert _scenario_delta(instruction) == expected + + def test_presentation_redaction_handles_prose_with_email_before_url(): prose = "Contact owner@example.com or see https://example.com/docs" @@ -124,6 +139,19 @@ def test_local_sandbox_reports_live_stage_timestamp_and_detail_from_events( "payload": {"stage": "scenarios"}, } ) + + "\n" + + json.dumps( + { + "type": "harness.stage.progress", + "wall_time": "2026-08-22T01:02:18+00:00", + "payload": { + "stage": "scenarios", + "detail": "Generating scenarios · validating 1/2", + "completed": 1, + "total": 2, + }, + } + ) + "\n", encoding="utf-8", ) @@ -131,8 +159,8 @@ def test_local_sandbox_reports_live_stage_timestamp_and_detail_from_events( current = client.get(f"/v1/jobs/{job_id}").json()["status"] assert current["stage"] == "generating_scenarios" - assert current["updated_at"] == "2026-08-22T01:02:03Z" - assert current["detail"] == "generating and validating scenarios" + assert current["updated_at"] == "2026-08-22T01:02:18Z" + assert current["detail"] == "Generating scenarios · validating 1/2" def test_local_sandbox_exposes_generated_stage_outputs_and_adjustments( @@ -459,6 +487,56 @@ async def submit(): assert reference.key.startswith("ALK_JOB_") +def test_controller_reporting_values_are_ephemeral_and_not_agent_runtime_values( + tmp_path, monkeypatch +): + source = tmp_path / "sources" / "agent" + source.mkdir(parents=True) + monkeypatch.setenv("ALK_SANDBOX_SOURCE_ROOTS", str(tmp_path / "sources")) + sandbox = LocalSandbox(tmp_path / "state") + observed = {} + + async def capture(job, _source): + observed["resolved"] = resolve_worker_secrets( + job.agent.secret_refs, + environment={ + **os.environ, + **sandbox._ephemeral_secrets[job.job_id], + }, + ) + observed["runtime_names"] = job.metadata["environment_value_names"] + + sandbox._execute = capture + + async def submit(): + response = sandbox.submit( + LocalSandboxRequest( + source_path=str(source), + controller_environment_values={ + "HARNESS_PLATFORM_API_KEY": "org-harness-key", + "HARNESS_PLATFORM_SECRET_KEY": "org-harness-secret", + "HARNESS_PLATFORM_WORKSPACE_ID": "workspace-1", + }, + ) + ) + await sandbox._tasks[response.job.job_id] + return response + + response = asyncio.run(submit()) + persisted = ( + tmp_path / "state" / "jobs" / response.job.job_id / "job.json" + ).read_text() + + assert observed["resolved"] == { + "HARNESS_PLATFORM_API_KEY": "org-harness-key", + "HARNESS_PLATFORM_SECRET_KEY": "org-harness-secret", + "HARNESS_PLATFORM_WORKSPACE_ID": "workspace-1", + } + assert observed["runtime_names"] == [] + assert "org-harness-key" not in persisted + assert "org-harness-secret" not in persisted + + def test_saved_session_rerun_reuses_artifacts_and_keeps_fresh_values_ephemeral( tmp_path, monkeypatch ): @@ -496,13 +574,20 @@ async def capture(job, only): observed["job"] = job.job_id observed["only"] = only observed["values"] = dict(sandbox._ephemeral_secrets[job.job_id]) + observed["controller"] = dict( + sandbox._ephemeral_controller_secrets[job.job_id] + ) sandbox._ephemeral_secrets.pop(job.job_id, None) + sandbox._ephemeral_controller_secrets.pop(job.job_id, None) sandbox._execute_rerun = capture queued = sandbox.rerun( job_id, SandboxRerunRequest( - environment_values={"OPENAI_API_KEY": "fresh-never-persist"} + environment_values={"OPENAI_API_KEY": "fresh-never-persist"}, + controller_environment_values={ + "HARNESS_PLATFORM_API_KEY": "org-key" + }, ), ) await sandbox._tasks[job_id] @@ -515,11 +600,13 @@ async def capture(job, only): "job": queued.job.job_id, "only": [], "values": {"OPENAI_API_KEY": "fresh-never-persist"}, + "controller": {"HARNESS_PLATFORM_API_KEY": "org-key"}, } assert "fresh-never-persist" not in state_path.read_text(encoding="utf-8") assert "fresh-never-persist" not in (state_path.parent / "job.json").read_text( encoding="utf-8" ) + assert "org-key" not in state_path.read_text(encoding="utf-8") def test_uploaded_environment_rejects_runner_control_and_reference_conflicts( diff --git a/tests/test_harness_service_environments.py b/tests/test_harness_service_environments.py index 82841586..86217089 100644 --- a/tests/test_harness_service_environments.py +++ b/tests/test_harness_service_environments.py @@ -525,6 +525,7 @@ def test_static_ports_receive_a_job_scoped_compose_override(tmp_path, monkeypatc assert "ports: !override" in generated assert 'published: "0"' in generated assert 'host_ip: "127.0.0.1"' in generated + assert 'com.futureagi.harness.project: "isolated"' in generated def test_fixed_container_name_is_reset_for_project_isolation(tmp_path): @@ -544,6 +545,28 @@ def test_fixed_container_name_is_reset_for_project_isolation(tmp_path): generated = Path(environment.compose_override_file).read_text() assert "container_name: !reset null" in generated + assert 'com.futureagi.harness.project: "isolated"' in generated + + +def test_missing_env_file_and_isolation_share_one_service_override(tmp_path): + compose = tmp_path / "compose.yml" + compose.write_text( + "services:\n agent:\n build: .\n env_file: .env.local\n" + ) + environment = ProvisionedEnvironment( + source=str(tmp_path), compose_file=str(compose), project="isolated" + ) + + _write_port_override( + tmp_path / "session", + environment, + {"services": {"agent": {"build": {"context": "."}}}}, + ) + + generated = Path(environment.compose_override_file).read_text() + assert generated.count(' "agent":') == 1 + assert "env_file: !reset []" in generated + assert 'com.futureagi.harness.project: "isolated"' in generated def test_profile_gated_service_ports_are_not_allocated(tmp_path, monkeypatch):