diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..b074a9e --- /dev/null +++ b/Makefile @@ -0,0 +1,22 @@ +.PHONY: help test test-nexus clean + +PYTHON ?= python3 +FRAMEWORK_DIR = nexus/framework + +help: + @echo "Available targets:" + @echo " make test Run all framework unit tests" + @echo " make test-nexus Run Nexus framework unit tests" + @echo " make clean Remove cache files and build artifacts" + +test: test-nexus + +test-nexus: + PYTHONPATH=$(FRAMEWORK_DIR) $(PYTHON) -m unittest discover -s $(FRAMEWORK_DIR)/tests -v + +clean: + find . -type d -name "__pycache__" -exec rm -rf {} + + find . -type f -name "*.pyc" -delete + find . -type f -name "*.pyo" -delete + find . -type d -name "*.egg-info" -exec rm -rf {} + + find . -type d -name ".pytest_cache" -exec rm -rf {} + diff --git a/nexus/framework/BUILD b/nexus/framework/BUILD index 647bed0..5a3b3e3 100644 --- a/nexus/framework/BUILD +++ b/nexus/framework/BUILD @@ -13,27 +13,94 @@ pytype_strict_library( ) pytype_strict_library( - name = "coworker", - srcs = ["coworker.py"], + name = "runtime", + srcs = ["runtime.py"], deps = [ ":utils", ], ) pytype_strict_binary( - name = "coworker_bin", - srcs = ["coworker.py"], - main = "coworker.py", + name = "runtime_bin", + srcs = ["runtime.py"], + main = "runtime.py", deps = [ - ":coworker", + ":runtime", + ], +) + +pytype_strict_library( + name = "compiler", + srcs = ["compiler.py"], + deps = [ + ":runtime", + ":utils", + ], +) + +pytype_strict_binary( + name = "compiler_bin", + srcs = ["compiler.py"], + main = "compiler.py", + deps = [ + ":compiler", + ], +) + +pytype_strict_library( + name = "installer", + srcs = ["installer.py"], + deps = [ + ":utils", + ], +) + +pytype_strict_binary( + name = "installer_bin", + srcs = ["installer.py"], + main = "installer.py", + deps = [ + ":installer", + ], +) + +pytype_strict_contrib_test( + name = "test_utils", + srcs = ["tests/test_utils.py"], + main = "tests/test_utils.py", + deps = [ + ":utils", + ], +) + +pytype_strict_contrib_test( + name = "test_runtime", + srcs = ["tests/test_runtime.py"], + main = "tests/test_runtime.py", + deps = [ + ":runtime", + ":utils", + ], +) + +pytype_strict_contrib_test( + name = "test_compiler", + srcs = ["tests/test_compiler.py"], + main = "tests/test_compiler.py", + deps = [ + ":compiler", + ":runtime", + ":utils", ], ) pytype_strict_contrib_test( - name = "test_coworker", - srcs = ["tests/test_coworker.py"], - main = "tests/test_coworker.py", + name = "test_installer", + srcs = ["tests/test_installer.py"], + main = "tests/test_installer.py", deps = [ - ":coworker", + ":compiler", + ":installer", + ":utils", ], ) diff --git a/nexus/framework/README.md b/nexus/framework/README.md index 3b3880a..3676af0 100644 --- a/nexus/framework/README.md +++ b/nexus/framework/README.md @@ -12,6 +12,17 @@ embedding domain-specific prompts or agent logic. --- +## Modular Architecture + +The framework is organized into three distinct, specialized tools: + +1. **`compiler.py`** *(Nexus/CI)*: Package verification, graph checks, and harness translation. +2. **`installer.py`** *(User Host)*: Distribution installation, environment setup, and uninstallation rollback. +3. **`runtime.py`** *(Target Workspace)*: Lightweight embedded runner for run namespaces, artifact descriptors, and message validation. +4. **`utils.py`** *(Shared)*: Common semver, JSON schema validation, safe path resolution, and configuration merging. + +--- + ## Package Lifecycle Workflow The canonical workflow comprises four core stages: **Verification**, @@ -31,7 +42,7 @@ compatibility targets, and delegation topologies (detecting circular delegations, depth exceeding two levels, and unreachable subagents): ```bash -python3 framework/coworker.py verify path/to/package +python3 framework/compiler.py verify path/to/package ``` ### 2. Harness Translation @@ -42,12 +53,12 @@ snapshot: ```bash # Compile for Claude Code -python3 framework/coworker.py translate path/to/package \ +python3 framework/compiler.py translate path/to/package \ --harness claude-code \ --output dist/claude-code/ # Compile for Codex -python3 framework/coworker.py translate path/to/package \ +python3 framework/compiler.py translate path/to/package \ --harness codex \ --output dist/codex/ ``` @@ -60,17 +71,17 @@ ledger (`ownership.json`): ```bash # Interactive installation -python3 framework/coworker.py install dist/codex/ \ +python3 framework/installer.py install dist/codex/ \ --destination /path/to/project # Non-interactive / Automated installation -python3 framework/coworker.py install dist/claude-code/ \ +python3 framework/installer.py install dist/claude-code/ \ --destination /path/to/project \ --answers path/to/answers.json \ --non-interactive # Upgrade an existing installation -python3 framework/coworker.py install dist/claude-code/ \ +python3 framework/installer.py install dist/claude-code/ \ --destination /path/to/project \ --upgrade ``` @@ -81,7 +92,7 @@ configuration entries while preserving user-created artifacts and unmodified settings: ```bash -python3 framework/coworker.py uninstall /path/to/project \ +python3 framework/installer.py uninstall /path/to/project \ --package ``` @@ -90,7 +101,7 @@ python3 framework/coworker.py uninstall /path/to/project \ ## Embedded Runtime Architecture Every generated distribution bundles a self-contained runtime snapshot under -`runtime/` (`coworker.py` and `utils.py`). Installed agent entrypoints invoke +`runtime/` (`runtime.py` and `utils.py`). Installed agent entrypoints invoke this local runtime to manage execution sessions and artifact lifecycles without relying on external framework installations. @@ -100,7 +111,7 @@ collision-resistant, timestamped run directory under `.coworker//runs//`: ```bash -python3 .coworker//runtime/coworker.py start-run \ +python3 .coworker//runtime/runtime.py start-run \ --workspace . \ --package ``` @@ -123,7 +134,7 @@ descriptors containing canonical URIs, content hashes, and schema metadata rather than embedding raw payloads across message boundaries: ```bash -python3 .coworker//runtime/coworker.py describe-artifact \ +python3 .coworker//runtime/runtime.py describe-artifact \ --workspace . \ --package \ --run-id \ @@ -149,7 +160,7 @@ Validates input and output instances against package schemas using the deterministic JSON Schema validator: ```bash -python3 .coworker//runtime/coworker.py validate \ +python3 .coworker//runtime/runtime.py validate \ --schema schemas/analysis-request.json \ instance.json ``` @@ -166,4 +177,4 @@ python3 .coworker//runtime/coworker.py validate \ 3. **Deterministic Verification**: Strict validation of delegation trees (no cycles, max depth <= 2, complete reachability) and semantic version ranges. 4. **Hermetic Runtime Snapshot**: Installed projects execute independently from - source trees via embedded `runtime/` bundles. + source trees via embedded `runtime/` bundles (`runtime.py`, `utils.py`). diff --git a/nexus/framework/compiler.py b/nexus/framework/compiler.py new file mode 100644 index 0000000..9dde85b --- /dev/null +++ b/nexus/framework/compiler.py @@ -0,0 +1,754 @@ +#!/usr/bin/env python3 +"""Compiler module for Coworker framework. + +Handles package manifest verification, structural integrity checks, target +harness compatibility validation, document generation, and compilation into +distribution formats (Claude Code and Codex). +""" + +from __future__ import annotations + +import argparse +import pathlib +import re +import shutil +import sys +from typing import Any, Callable, Sequence + +try: + from accelerator_agents.tpu_nexus.framework import utils +except ImportError: + try: + from . import utils + except ImportError: + import utils + +Path = pathlib.Path + +CoworkerError = utils.CoworkerError +FRAMEWORK_VERSION = utils.FRAMEWORK_VERSION +GENERATED_MARKER = utils.GENERATED_MARKER +PORTABLE_TOOLS = utils.PORTABLE_TOOLS +load_json = utils.load_json +parse_version_range = utils.parse_version_range +rooted = utils.rooted +safe_relative = utils.safe_relative +validate_package_name = utils.validate_package_name +validate_schema_definition = utils.validate_schema_definition +write_json = utils.write_json + + +# ===================================================================== +# Package Validation & Graph Checking +# ===================================================================== + + +def render_agent_references( + text: str, manifest: dict[str, Any], prefix: str = "" +) -> str: + """Replace {{agent:NAME}} template tokens with prefixed agent identifiers.""" + known_agents = {agent["name"] for agent in manifest["agents"]} + + def replace(match: re.Match[str]) -> str: + name = match.group(1) + if name not in known_agents: + raise CoworkerError(f"instructions reference unknown agent: {name}") + return prefix + name + + rendered = re.sub(r"\{\{agent:([a-z0-9]+(?:-[a-z0-9]+)*)}}", replace, text) + if "{{agent:" in rendered: + raise CoworkerError("instructions contain a malformed agent reference") + return rendered + + +def validate_delegation_graph( + entrypoint: dict[str, Any], + agents_by_name: dict[str, dict[str, Any]], + expected_agents: set[str], +) -> None: + """Validate delegation graph for cycle detection, max depth <= 2, and reachability.""" + visited: set[str] = set() + + def visit(agent_name: str, depth: int, path: tuple[str, ...]) -> None: + if agent_name in path: + cycle = " -> ".join((*path, agent_name)) + raise CoworkerError(f"delegation graph contains a cycle: {cycle}") + if depth > 2: + chain = " -> ".join((*path, agent_name)) + raise CoworkerError( + f"delegation exceeds two levels below entrypoint: {chain}" + ) + + visited.add(agent_name) + for delegated in agents_by_name[agent_name].get("delegates", []): + visit(delegated, depth + 1, (*path, agent_name)) + + for delegated in entrypoint["delegates"]: + visit(delegated, 1, (entrypoint["name"],)) + + unreachable = expected_agents - visited + if unreachable: + raise CoworkerError( + f"agents are unreachable from entrypoint: {sorted(unreachable)}" + ) + + +def load_package(package_dir: Path) -> dict[str, Any]: + """Load, validate, and verify the structural integrity of a package manifest and its graph.""" + manifest_path = package_dir / "package.json" + manifest = load_json(manifest_path) + + # 1. Manifest level requirements + required_manifest_fields = { + "name", + "version", + "description", + "entrypoint", + "agents", + "environment", + "compatibility_targets", + } + missing_manifest = required_manifest_fields - manifest.keys() + if missing_manifest: + raise CoworkerError( + f"package.json missing: {', '.join(sorted(missing_manifest))}" + ) + manifest_name = manifest["name"] + + validate_package_name(manifest_name) + + # 2. Agent presence & uniqueness + agent_names = [agent.get("name") for agent in manifest["agents"]] + if None in agent_names or len(agent_names) != len(set(agent_names)): + raise CoworkerError( + f"Agent names are not unique in package or are missing: {manifest_name}" + ) + name_set = set(agent_names) + + # 3. Entrypoint validation + entrypoint = manifest["entrypoint"] + entry_point_is_missing = { + "name", + "instructions", + "delegates", + } - entrypoint.keys() + if entry_point_is_missing: + raise CoworkerError( + f"Entrypoint {manifest_name} missing fields:" + f" {sorted(entry_point_is_missing)}" + ) + if entrypoint["name"] in name_set: + raise CoworkerError("Entrypoint and agent names must be distinct") + + unknown_entry_delegates = set(entrypoint["delegates"]) - name_set + if unknown_entry_delegates: + raise CoworkerError( + "entrypoint delegates to unknown agents:" + f" {sorted(unknown_entry_delegates)}" + ) + + # 4. Agent definitions & files + agents_by_name = {agent["name"]: agent for agent in manifest["agents"]} + for agent in manifest["agents"]: + required_agent_fields = { + "name", + "description", + "instructions", + "accepts", + "produces", + "delegates", + "tools", + } + missing_agent = required_agent_fields - agent.keys() + if missing_agent: + agent_label = agent.get("name", "") + raise CoworkerError( + f"agent {agent_label} missing: {sorted(missing_agent)}" + ) + + unknown_tools = set(agent["tools"]) - PORTABLE_TOOLS + if unknown_tools: + raise CoworkerError( + f"agent {agent['name']} has unsupported portable tools:" + f" {sorted(unknown_tools)}" + ) + + unknown_delegates = set(agent.get("delegates", [])) - name_set + if unknown_delegates: + raise CoworkerError( + f"agent {agent['name']} delegates to unknown agents:" + f" {unknown_delegates}" + ) + + source_inst = rooted( + package_dir, agent["instructions"], "agent instructions" + ) + if not source_inst.is_file(): + raise CoworkerError(f"missing agent instructions: {source_inst}") + render_agent_references( + source_inst.read_text(encoding="utf-8"), manifest, "" + ) + + for schema_key in ("accepts", "produces"): + schema_path = rooted( + package_dir, agent[schema_key], f"agent {schema_key}" + ) + if not schema_path.is_file(): + raise CoworkerError(f"missing agent {schema_key} schema: {schema_path}") + + # 5. Delegation graph traversal: cycle detection, max depth <= 2, reachability + validate_delegation_graph(entrypoint, agents_by_name, name_set) + + # 6. Entrypoint and environment files verification + entry_inst = rooted( + package_dir, entrypoint["instructions"], "entrypoint instructions" + ) + if not entry_inst.is_file(): + raise CoworkerError(f"missing entrypoint instructions: {entry_inst}") + render_agent_references(entry_inst.read_text(encoding="utf-8"), manifest, "") + + questions = rooted( + package_dir, manifest["environment"]["questions"], "environment questions" + ) + env_schema = rooted( + package_dir, manifest["environment"]["schema"], "environment schema" + ) + if not questions.is_file() or not env_schema.is_file(): + raise CoworkerError("environment questions or schema is missing") + + return manifest + + +def target_for(manifest: dict[str, Any], harness: str) -> dict[str, Any]: + """Retrieve and validate the compatibility target configuration for a given harness.""" + target = next( + ( + t + for t in manifest.get("compatibility_targets", []) + if t.get("harness") == harness + ), + None, + ) + if not target: + raise CoworkerError(f"no compatibility target for harness {harness}") + + # Validate version constraint syntax + parse_version_range(target["versions"]) + + # Verify target satisfies all required capabilities + required = set(manifest.get("required_capabilities", [])) + available = {k for k, v in target.get("capabilities", {}).items() if v} + missing = required - available + if missing: + raise CoworkerError( + f"target {target['name']} cannot preserve required capabilities:" + f" {', '.join(sorted(missing))}" + ) + + return target + + +def verify_package(package_dir: Path) -> None: + """Verify package manifest, schema definitions, and target compatibility.""" + manifest = load_package(package_dir) + for schema_rel in manifest.get("schemas", []): + schema_path = rooted(package_dir, schema_rel, "schema path") + schema = load_json(schema_path) + errors = validate_schema_definition(schema) + if errors: + raise CoworkerError(f"invalid schema {schema_rel}:\n" + "\n".join(errors)) + + harnesses = { + target["harness"] for target in manifest["compatibility_targets"] + } + for harness in harnesses: + target_for(manifest, harness) + + +# ===================================================================== +# Document & Prompt Generators +# ===================================================================== + + +def generated_header(manifest: dict[str, Any], target: dict[str, Any]) -> str: + """Generate the standard comment header for translated assets.""" + return ( + f"\n\n" + ) + + +def skill_frontmatter(name: str, description: str) -> str: + """Generate YAML frontmatter for an entrypoint skill.""" + escaped = description.replace('"', "'") + return f'---\nname: {name}\ndescription: "{escaped}"\n---\n\n' + + +def skill_package_contract(package_name: str, delegates: list[str]) -> str: + """Build the runtime execution contract section for an installed skill package.""" + targets = ", ".join(delegates) or "none" + runtime = f"python3 .coworker/{package_name}/runtime/runtime.py" + return f""" + +## Installed package contract + +- Read `.coworker/{package_name}/environment.json` before delegating. +- Before the first delegation, run `{runtime} start-run --workspace . --package {package_name}` exactly once. +- Retain the returned `run_id`, include it in every delegation request, and use only that run's artifact references. +- Direct delegation targets: {targets}. Do not invoke any other package agent directly. +- Run one branch at a time. A subagent may wait for its one nested helper; do not run sibling branches in parallel. +- Validate messages with `{runtime} validate`. +- Store durable outputs below `.coworker/{package_name}/runs//artifacts/`; never overwrite an existing artifact. +- Create descriptors with `{runtime} describe-artifact --workspace . --package {package_name} --run-id --file --schema --media-type ` and pass descriptors, not payload copies. +""" + + +def skill_document( + package_dir: Path, + manifest: dict[str, Any], + target: dict[str, Any], + agent_prefix: str = "", +) -> str: + """Generate SKILL.md markdown document for entrypoint workflow.""" + entry = manifest["entrypoint"] + frontmatter = skill_frontmatter(entry["name"], manifest["description"]) + header = generated_header(manifest, target) + + source_path = rooted( + package_dir, entry["instructions"], "entrypoint instructions" + ) + rendered_body = render_agent_references( + source_path.read_text(encoding="utf-8"), manifest, agent_prefix + ).rstrip() + + delegates = [agent_prefix + name for name in entry["delegates"]] + contract = skill_package_contract(manifest["name"], delegates) + + return f"{frontmatter}{header}{rendered_body}{contract}" + + +def agent_execution_contract( + package_name: str, agent: dict[str, Any], delegates: list[str] +) -> str: + """Build the runtime execution contract section for an agent prompt.""" + targets = ", ".join(delegates) or "none" + runtime = f"python3 .coworker/{package_name}/runtime/runtime.py" + return f""" + +## Generated execution contract + +- Validate the request against `.coworker/{package_name}/{agent['accepts']}` before work. +- Require the request's `run_id` and keep every output under `.coworker/{package_name}/runs//artifacts/`. +- Validate the result against `.coworker/{package_name}/{agent['produces']}` before returning. +- Use `{runtime} validate --schema SCHEMA INSTANCE`. +- Return the common result envelope with `completed`, `invalid_input`, `needs_input`, or `failed`. +- Never overwrite an existing artifact; create its descriptor with `{runtime} describe-artifact --workspace . --package {package_name} --run-id --file --schema --media-type `. +- Reject artifact references whose URI does not contain the request's exact package and `run_id`. +- Materialize durable outputs and return artifact descriptors; do not copy payloads into messages. +- Allowed delegation targets: {targets}. +- Run sequentially. Never weaken inherited permissions or approvals. +""" + + +def agent_prompt( + package_dir: Path, + manifest: dict[str, Any], + target: dict[str, Any], + agent: dict[str, Any], + agent_prefix: str = "", +) -> str: + """Generate agent instruction prompt document with generated execution contract.""" + source_path = rooted(package_dir, agent["instructions"], "agent instructions") + rendered_body = render_agent_references( + source_path.read_text(encoding="utf-8"), manifest, agent_prefix + ).rstrip() + + delegates = [agent_prefix + name for name in agent.get("delegates", [])] + contract = agent_execution_contract(manifest["name"], agent, delegates) + + return f"{generated_header(manifest, target)}{rendered_body}{contract}" + + +def claude_frontmatter(agent: dict[str, Any]) -> str: + """Generate Claude agent YAML frontmatter with mapped native tools.""" + tool_map = { + "read": "Read", + "search": "Grep, Glob", + "shell": "Bash", + "write": "Write", + "edit": "Edit", + } + tools: list[str] = [] + for portable in agent["tools"]: + for native in tool_map[portable].split(", "): + if native not in tools: + tools.append(native) + if agent.get("delegates"): + tools.append("Agent") + + return ( + "---\n" + f"name: {agent['name']}\n" + f"description: {agent['description']}\n" + f"tools: {', '.join(tools)}\n" + "background: false\n" + "---\n\n" + ) + + +# ===================================================================== +# Harness Translation +# ===================================================================== + + +def copy_asset(source: Path, destination: Path) -> None: + """Copy a file or directory tree to destination, creating parent directories as needed.""" + if source.is_dir(): + shutil.copytree(source, destination, dirs_exist_ok=True) + elif source.is_file(): + destination.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, destination) + else: + raise CoworkerError(f"copy path does not exist: {source}") + + +def bundle_runtime_scripts(runtime_dir: Path) -> None: + """Bundle framework runtime worker scripts into the distribution runtime directory.""" + runtime_dir.mkdir(parents=True, exist_ok=True) + framework_dir = Path(__file__).resolve().parent + + for script_name in ("runtime.py", "utils.py"): + script_path = framework_dir / script_name + if script_path.is_file(): + shutil.copy2(script_path, runtime_dir / script_name) + + +def copy_runtime_assets( + package_dir: Path, output: Path, manifest: dict[str, Any] +) -> None: + """Copy configured package assets and bundle framework runtime worker scripts.""" + for item in manifest.get("copy", []): + rel = safe_relative(item, "copy path") + source = rooted(package_dir, str(rel), "copy path") + copy_asset(source, output / rel) + + bundle_runtime_scripts(output / "runtime") + + +def runtime_copy_mappings( + package_name: str, copy_items: Sequence[str] +) -> list[dict[str, str]]: + """Build list of runtime and schema copy mappings for an installed package.""" + mappings = [ + {"source": "runtime", "destination": f".coworker/{package_name}/runtime"}, + {"source": "schemas", "destination": f".coworker/{package_name}/schemas"}, + ] + if any(Path(item).parts[:1] == ("scripts",) for item in copy_items): + mappings.append({ + "source": "scripts", + "destination": f".coworker/{package_name}/scripts", + }) + mappings.append({ + "source": "coworker-build.json", + "destination": f".coworker/{package_name}/coworker-build.json", + }) + return mappings + + +def harness_install_specs(entry_name: str) -> dict[str, dict[str, Any]]: + """Return harness-specific copy rules, version command, and configuration settings.""" + return { + "claude-code": { + "version_command": ["claude", "--version"], + "copies": [ + { + "source": f"project/skills/{entry_name}", + "destination": f".claude/skills/{entry_name}", + }, + {"source": "project/agents", "destination": ".claude/agents"}, + ], + "json_merges": [{ + "path": ".claude/settings.local.json", + "value": { + "env": { + "CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH": "2", + "CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": "2", + } + }, + }], + "toml_sets": [], + }, + "codex": { + "version_command": ["codex", "--version"], + "copies": [ + { + "source": f"skills/{entry_name}", + "destination": f".agents/skills/{entry_name}", + }, + {"source": "agents", "destination": ".codex/agents"}, + ], + "json_merges": [], + "toml_sets": [ + { + "path": ".codex/config.toml", + "section": "agents", + "key": "enabled", + "value": True, + }, + { + "path": ".codex/config.toml", + "section": "agents", + "key": "max_concurrent_threads_per_session", + "value": 2, + }, + ], + }, + } + + +def build_install_plan( + manifest: dict[str, Any], + target: dict[str, Any], + harness: str, +) -> dict[str, Any]: + """Generate coworker-install.json installation plan for target harness.""" + name = manifest["name"] + entry = manifest["entrypoint"]["name"] + spec = harness_install_specs(entry)[harness] + + return { + "harness": harness, + "version_range": target["versions"], + "environment_questions": "environment/questions.json", + "environment_schema": manifest["environment"]["schema"], + "environment_path": f".coworker/{name}/environment.json", + "ownership_path": f".coworker/{name}/ownership.json", + "version_command": spec["version_command"], + "copies": ( + spec["copies"] + runtime_copy_mappings(name, manifest.get("copy", [])) + ), + "json_merges": spec["json_merges"], + "toml_sets": spec["toml_sets"], + } + + +def translate_claude( + package_dir: Path, + output: Path, + manifest: dict[str, Any], + target: dict[str, Any], +) -> None: + """Translate package into Claude Code plugin and project layouts.""" + plugin = output / ".claude-plugin" + write_json( + plugin / "plugin.json", + { + "name": manifest["name"], + "displayName": manifest.get("display_name", manifest["name"]), + "version": manifest["version"], + "description": manifest["description"], + "author": {"name": manifest.get("author", "Coworker Framework")}, + }, + ) + + entry_name = manifest["entrypoint"]["name"] + plugin_agent_prefix = f"{manifest['name']}:" + + # Write plugin skill and standalone project skill + skill = output / "skills" / entry_name / "SKILL.md" + skill.parent.mkdir(parents=True, exist_ok=True) + skill.write_text( + skill_document(package_dir, manifest, target, plugin_agent_prefix), + encoding="utf-8", + ) + + project_skill = output / "project" / "skills" / entry_name / "SKILL.md" + project_skill.parent.mkdir(parents=True, exist_ok=True) + project_skill.write_text( + skill_document(package_dir, manifest, target), encoding="utf-8" + ) + + # Write plugin agents and project agents + for agent in manifest["agents"]: + fm = claude_frontmatter(agent) + path = output / "agents" / f"{agent['name']}.md" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + fm + + agent_prompt( + package_dir, manifest, target, agent, plugin_agent_prefix + ), + encoding="utf-8", + ) + + project_agent = output / "project" / "agents" / f"{agent['name']}.md" + project_agent.parent.mkdir(parents=True, exist_ok=True) + project_agent.write_text( + fm + agent_prompt(package_dir, manifest, target, agent), + encoding="utf-8", + ) + + write_json( + output / "coworker-install.json", + build_install_plan(manifest, target, "claude-code"), + ) + + +def translate_codex( + package_dir: Path, + output: Path, + manifest: dict[str, Any], + target: dict[str, Any], +) -> None: + """Translate package into Codex plugin and agents layouts.""" + plugin = output / ".codex-plugin" + write_json( + plugin / "plugin.json", + { + "name": manifest["name"], + "version": manifest["version"], + "description": manifest["description"], + "author": {"name": manifest.get("author", "Coworker Framework")}, + "skills": "./skills/", + "interface": { + "displayName": manifest.get("display_name", manifest["name"]), + "shortDescription": manifest["description"][:80], + "longDescription": manifest["description"], + "developerName": manifest.get("author", "Coworker Framework"), + "category": "Productivity", + "capabilities": ["Interactive", "Write"], + "defaultPrompt": [ + manifest["entrypoint"].get( + "example_prompt", "Run this workflow." + )[:128] + ], + }, + }, + ) + + entry_name = manifest["entrypoint"]["name"] + skill = output / "skills" / entry_name / "SKILL.md" + skill.parent.mkdir(parents=True, exist_ok=True) + skill.write_text( + skill_document(package_dir, manifest, target), encoding="utf-8" + ) + + for agent in manifest["agents"]: + config = output / "agents" / f"{agent['name']}.toml" + config.parent.mkdir(parents=True, exist_ok=True) + prompt = agent_prompt(package_dir, manifest, target, agent).replace( + '"""', "'''" + ) + sandbox = ( + "workspace-write" + if any(tool in agent["tools"] for tool in ("write", "edit")) + else "read-only" + ) + config.write_text( + f'name = "{agent["name"]}"\n' + f'description = "{agent["description"]}"\n' + f'sandbox_mode = "{sandbox}"\n' + f'developer_instructions = """\n{prompt}\n"""\n', + encoding="utf-8", + ) + + write_json( + output / "coworker-install.json", + build_install_plan(manifest, target, "codex"), + ) + + +def translate(package_dir: Path, harness: str, output: Path) -> None: + """Translate a package into a target harness distribution.""" + manifest = load_package(package_dir) + target = target_for(manifest, harness) + + if output.exists(): + marker = output / "coworker-build.json" + if not marker.is_file() or not load_json(marker).get("generated"): + raise CoworkerError( + f"refusing to replace non-generated output directory: {output}" + ) + shutil.rmtree(output) + + output.mkdir(parents=True) + + translators: dict[ + str, Callable[[Path, Path, dict[str, Any], dict[str, Any]], None] + ] = { + "claude-code": translate_claude, + "codex": translate_codex, + } + translators[harness](package_dir, output, manifest, target) + + copy_runtime_assets(package_dir, output, manifest) + + write_json( + output / "coworker-build.json", + { + "framework_version": FRAMEWORK_VERSION, + "source": manifest["name"], + "source_version": manifest["version"], + "compatibility_target": target["name"], + "harness": harness, + "version_range": target["versions"], + "generated": True, + }, + ) + + +def build_compiler_parser() -> argparse.ArgumentParser: + """Construct parser for standalone compiler commands.""" + parser = argparse.ArgumentParser( + prog="coworker-compiler", + description="Verification and translation compiler for Coworker packages.", + ) + sub = parser.add_subparsers(dest="command", required=True) + + # verify + p_verify = sub.add_parser( + "verify", help="Verify package structure and schemas" + ) + p_verify.add_argument("package", type=Path, help="Path to package directory") + + # translate + p_translate = sub.add_parser( + "translate", help="Translate package to target harness" + ) + p_translate.add_argument( + "package", type=Path, help="Path to package directory" + ) + p_translate.add_argument( + "--harness", + choices=["claude-code", "codex"], + required=True, + help="Target harness", + ) + p_translate.add_argument( + "--output", type=Path, required=True, help="Output directory" + ) + + return parser + + +def main(argv: list[str] | None = None) -> int: + """Main entrypoint for standalone compiler operations.""" + parser = build_compiler_parser() + args = parser.parse_args(argv) + + try: + if args.command == "verify": + verify_package(args.package) + + elif args.command == "translate": + translate(args.package, args.harness, args.output) + + return 0 + except CoworkerError as exc: + print(f"compiler: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/nexus/framework/coworker.py b/nexus/framework/coworker.py deleted file mode 100644 index 434c38e..0000000 --- a/nexus/framework/coworker.py +++ /dev/null @@ -1,1611 +0,0 @@ -#!/usr/bin/env python3 -"""Harness-agnostic framework for running agents. - -This tool compiles, validates, installs, and manages canonical agent packages -across target agent harnesses (such as Claude Code and Codex) while providing -runtime support for run namespace isolation and artifact provenance. -""" - -from __future__ import annotations - -import argparse -import datetime -import pathlib -import re -import shutil -import subprocess -import sys -from typing import Any, Callable, Sequence -import uuid - -from nexus.framework import utils - -Path = pathlib.Path - -# ===================================================================== -# Utilities, Versioning, Schema Validation, & Config Managers -# ===================================================================== - -CoworkerError = utils.CoworkerError -FRAMEWORK_VERSION = utils.FRAMEWORK_VERSION -GENERATED_MARKER = utils.GENERATED_MARKER -JsonConfigManager = utils.JsonConfigManager -PORTABLE_TOOLS = utils.PORTABLE_TOOLS -SemVer = utils.SemVer -TomlConfigManager = utils.TomlConfigManager -compute_digest = utils.compute_digest -format_json = utils.format_json -load_json = utils.load_json -parse_version_range = utils.parse_version_range -rooted = utils.rooted -safe_relative = utils.safe_relative -validate_instance = utils.validate_instance -validate_package_name = utils.validate_package_name -validate_run_id = utils.validate_run_id -validate_schema_definition = utils.validate_schema_definition -version_satisfies = utils.version_satisfies -write_json = utils.write_json - -# ===================================================================== -# Package Validation & Graph Checking -# ===================================================================== - - -def render_agent_references( - text: str, manifest: dict[str, Any], prefix: str = "" -) -> str: - """Replace {{agent:NAME}} template tokens with prefixed agent identifiers.""" - known_agents = {agent["name"] for agent in manifest["agents"]} - - def replace(match: re.Match[str]) -> str: - name = match.group(1) - if name not in known_agents: - raise CoworkerError(f"instructions reference unknown agent: {name}") - return prefix + name - - rendered = re.sub(r"\{\{agent:([a-z0-9]+(?:-[a-z0-9]+)*)}}", replace, text) - if "{{agent:" in rendered: - raise CoworkerError("instructions contain a malformed agent reference") - return rendered - - -def validate_delegation_graph( - entrypoint: dict[str, Any], - agents_by_name: dict[str, dict[str, Any]], - expected_agents: set[str], -) -> None: - """Traverse delegation graph asserting max depth <= 2 below entrypoint, no cycles, and reachability.""" - visited: set[str] = set() - - def visit(agent_name: str, depth: int, path: tuple[str, ...]) -> None: - if agent_name in path: - cycle = " -> ".join((*path, agent_name)) - raise CoworkerError(f"delegation cycle detected: {cycle}") - if depth > 2: - chain = " -> ".join((*path, agent_name)) - raise CoworkerError( - f"delegation exceeds two levels below entrypoint: {chain}" - ) - - visited.add(agent_name) - for delegated in agents_by_name[agent_name].get("delegates", []): - visit(delegated, depth + 1, (*path, agent_name)) - - for delegated in entrypoint["delegates"]: - visit(delegated, 1, (entrypoint["name"],)) - - unreachable = expected_agents - visited - if unreachable: - raise CoworkerError( - f"agents are unreachable from entrypoint: {sorted(unreachable)}" - ) - - -def assert_no_yaml_frontmatter(content: str, filename: str, label: str) -> None: - """Assert that an instruction document content does not contain YAML frontmatter.""" - if content.lstrip().startswith("---"): - raise CoworkerError( - f"{label} file '{filename}' must not contain YAML frontmatter" - " (name and description belong in package.json)" - ) - - -def load_package(package_dir: Path) -> dict[str, Any]: - """Load, validate, and verify the structural integrity of a package manifest and its graph.""" - manifest_path = package_dir / "package.json" - manifest = utils.load_json(manifest_path) - - # 1. Manifest level requirements - required_manifest_fields = { - "name", - "version", - "description", - "entrypoint", - "agents", - "environment", - "compatibility_targets", - } - missing_manifest = required_manifest_fields - manifest.keys() - if missing_manifest: - raise CoworkerError( - f"package.json missing: {', '.join(sorted(missing_manifest))}" - ) - manifest_name = manifest["name"] - - validate_package_name(manifest_name) - - # 2. Agent presence & uniqueness - agent_names = [agent.get("name") for agent in manifest["agents"]] - if None in agent_names or len(agent_names) != len(set(agent_names)): - raise CoworkerError( - f"Agent names are not unique in package or are missing: {manifest_name}" - ) - name_set = set(agent_names) - - # 3. Entrypoint validation - entrypoint = manifest["entrypoint"] - entry_point_is_missing = { - "name", - "instructions", - "delegates", - } - entrypoint.keys() - if entry_point_is_missing: - raise CoworkerError( - f"Entrypoint {manifest_name} missing fields:" - f" {sorted(entry_point_is_missing)}" - ) - if entrypoint["name"] in name_set: - raise CoworkerError("Entrypoint and agent names must be distinct") - - unknown_entry_delegates = set(entrypoint["delegates"]) - name_set - if unknown_entry_delegates: - raise CoworkerError( - "entrypoint delegates to unknown agents:" - f" {sorted(unknown_entry_delegates)}" - ) - - # 4. Agent definitions & files - agents_by_name = {agent["name"]: agent for agent in manifest["agents"]} - for agent in manifest["agents"]: - required_agent_fields = { - "name", - "description", - "instructions", - "accepts", - "produces", - "delegates", - "tools", - } - missing_agent = required_agent_fields - agent.keys() - if missing_agent: - agent_label = agent.get("name", "") - raise CoworkerError( - f"agent {agent_label} missing: {sorted(missing_agent)}" - ) - - unknown_tools = set(agent["tools"]) - utils.PORTABLE_TOOLS - if unknown_tools: - raise CoworkerError( - f"agent {agent['name']} has unsupported portable tools:" - f" {sorted(unknown_tools)}" - ) - - unknown_delegates = set(agent.get("delegates", [])) - name_set - if unknown_delegates: - raise CoworkerError( - f"agent {agent['name']} delegates to unknown agents:" - f" {unknown_delegates}" - ) - - source_inst = utils.rooted( - package_dir, agent["instructions"], "agent instructions" - ) - if not source_inst.is_file(): - raise CoworkerError(f"missing agent instructions: {source_inst}") - inst_content = source_inst.read_text(encoding="utf-8") - assert_no_yaml_frontmatter(inst_content, source_inst.name, "agent instructions") - render_agent_references(inst_content, manifest, "") - - for schema_key in ("accepts", "produces"): - schema_path = utils.rooted( - package_dir, agent[schema_key], f"agent {schema_key}" - ) - if not schema_path.is_file(): - raise CoworkerError(f"missing agent {schema_key} schema: {schema_path}") - - # 5. Delegation graph traversal: cycle detection, max depth <= 2, reachability - validate_delegation_graph(entrypoint, agents_by_name, name_set) - - # 6. Entrypoint and environment files verification - entry_inst = utils.rooted( - package_dir, entrypoint["instructions"], "entrypoint instructions" - ) - if not entry_inst.is_file(): - raise CoworkerError(f"missing entrypoint instructions: {entry_inst}") - entry_content = entry_inst.read_text(encoding="utf-8") - assert_no_yaml_frontmatter(entry_content, entry_inst.name, "entrypoint instructions") - render_agent_references(entry_content, manifest, "") - - questions = utils.rooted( - package_dir, manifest["environment"]["questions"], "environment questions" - ) - env_schema = utils.rooted( - package_dir, manifest["environment"]["schema"], "environment schema" - ) - if not questions.is_file() or not env_schema.is_file(): - raise CoworkerError("environment questions or schema is missing") - - return manifest - - -def target_for(manifest: dict[str, Any], harness: str) -> dict[str, Any]: - """Retrieve and validate the compatibility target configuration for a given harness.""" - target = next( - ( - t - for t in manifest.get("compatibility_targets", []) - if t.get("harness") == harness - ), - None, - ) - if not target: - raise CoworkerError(f"no compatibility target for harness {harness}") - - # Validate version constraint syntax - utils.parse_version_range(target["versions"]) - - # Verify target satisfies all required capabilities - required = set(manifest.get("required_capabilities", [])) - available = {k for k, v in target.get("capabilities", {}).items() if v} - missing = required - available - if missing: - raise CoworkerError( - f"target {target['name']} cannot preserve required capabilities:" - f" {', '.join(sorted(missing))}" - ) - - return target - - -def verify_package(package_dir: Path) -> dict[str, Any]: - """Verify package manifest, schema definitions, and target compatibility.""" - manifest = load_package(package_dir) - for schema_rel in manifest.get("schemas", []): - schema_path = utils.rooted(package_dir, schema_rel, "schema path") - schema = utils.load_json(schema_path) - errors = validate_schema_definition(schema) - if errors: - raise CoworkerError(f"invalid schema {schema_rel}:\n" + "\n".join(errors)) - - harnesses = { - target["harness"] for target in manifest["compatibility_targets"] - } - for harness in harnesses: - target_for(manifest, harness) - - return manifest - - -# ===================================================================== -# Document & Prompt Generators -# ===================================================================== - - -def generated_header(manifest: dict[str, Any], target: dict[str, Any]) -> str: - """Generate the standard comment header for translated assets.""" - return ( - f"\n\n" - ) - - -def skill_frontmatter(name: str, description: str) -> str: - """Generate YAML frontmatter for an entrypoint skill.""" - escaped = description.replace('"', "'") - return f'---\nname: {name}\ndescription: "{escaped}"\n---\n\n' - - -def skill_package_contract(package_name: str, delegates: list[str]) -> str: - """Build the runtime execution contract section for an installed skill package.""" - targets = ", ".join(delegates) or "none" - runtime = f"python3 .coworker/{package_name}/runtime/coworker.py" - return f""" - -## Installed package contract - -- Read `.coworker/{package_name}/environment.json` before delegating. -- Before the first delegation, run `{runtime} start-run --workspace . --package {package_name}` exactly once. -- Retain the returned `run_id`, include it in every delegation request, and use only that run's artifact references. -- Direct delegation targets: {targets}. Do not invoke any other package agent directly. -- Run one branch at a time. A subagent may wait for its one nested helper; do not run sibling branches in parallel. -- Validate messages with `{runtime} validate`. -- Store durable outputs below `.coworker/{package_name}/runs//artifacts/`; never overwrite an existing artifact. -- Create descriptors with `{runtime} describe-artifact --workspace . --package {package_name} --run-id --file --schema --media-type ` and pass descriptors, not payload copies. -""" - - -def skill_document( - package_dir: Path, - manifest: dict[str, Any], - target: dict[str, Any], - agent_prefix: str = "", -) -> str: - """Generate SKILL.md markdown document for entrypoint workflow.""" - entry = manifest["entrypoint"] - frontmatter = skill_frontmatter(entry["name"], manifest["description"]) - header = generated_header(manifest, target) - - source_path = rooted( - package_dir, entry["instructions"], "entrypoint instructions" - ) - rendered_body = render_agent_references( - source_path.read_text(encoding="utf-8"), manifest, agent_prefix - ).rstrip() - - delegates = [agent_prefix + name for name in entry["delegates"]] - contract = skill_package_contract(manifest["name"], delegates) - - return f"{frontmatter}{header}{rendered_body}{contract}" - - -def agent_execution_contract( - package_name: str, agent: dict[str, Any], delegates: list[str] -) -> str: - """Build the runtime execution contract section for an agent prompt.""" - targets = ", ".join(delegates) or "none" - runtime = f"python3 .coworker/{package_name}/runtime/coworker.py" - return f""" - -## Generated execution contract - -- Validate the request against `.coworker/{package_name}/{agent['accepts']}` before work. -- Require the request's `run_id` and keep every output under `.coworker/{package_name}/runs//artifacts/`. -- Validate the result against `.coworker/{package_name}/{agent['produces']}` before returning. -- Use `{runtime} validate --schema SCHEMA INSTANCE`. -- Return the common result envelope with `completed`, `invalid_input`, `needs_input`, or `failed`. -- Never overwrite an existing artifact; create its descriptor with `{runtime} describe-artifact --workspace . --package {package_name} --run-id --file --schema --media-type `. -- Reject artifact references whose URI does not contain the request's exact package and `run_id`. -- Materialize durable outputs and return artifact descriptors; do not copy payloads into messages. -- Allowed delegation targets: {targets}. -- Run sequentially. Never weaken inherited permissions or approvals. -""" - - -def agent_prompt( - package_dir: Path, - manifest: dict[str, Any], - target: dict[str, Any], - agent: dict[str, Any], - agent_prefix: str = "", -) -> str: - """Generate agent instruction prompt document with generated execution contract.""" - source_path = rooted(package_dir, agent["instructions"], "agent instructions") - rendered_body = render_agent_references( - source_path.read_text(encoding="utf-8"), manifest, agent_prefix - ).rstrip() - - delegates = [agent_prefix + name for name in agent.get("delegates", [])] - contract = agent_execution_contract(manifest["name"], agent, delegates) - - return f"{generated_header(manifest, target)}{rendered_body}{contract}" - - -def claude_frontmatter(agent: dict[str, Any]) -> str: - """Generate Claude agent YAML frontmatter with mapped native tools.""" - tool_map = { - "read": "Read", - "search": "Grep, Glob", - "shell": "Bash", - "write": "Write", - "edit": "Edit", - } - tools: list[str] = [] - for portable in agent["tools"]: - for native in tool_map[portable].split(", "): - if native not in tools: - tools.append(native) - if agent.get("delegates"): - tools.append("Agent") - - return ( - "---\n" - f"name: {agent['name']}\n" - f"description: {agent['description']}\n" - f"tools: {', '.join(tools)}\n" - "background: false\n" - "---\n\n" - ) - - -# ===================================================================== -# Harness Translation -# ===================================================================== - - -def copy_asset(source: Path, destination: Path) -> None: - """Copy a file or directory tree to destination, creating parent directories as needed.""" - if source.is_dir(): - shutil.copytree(source, destination, dirs_exist_ok=True) - elif source.is_file(): - destination.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source, destination) - else: - raise CoworkerError(f"copy path does not exist: {source}") - - -def bundle_runtime_scripts(runtime_dir: Path) -> None: - """Bundle framework runtime worker scripts into the distribution runtime directory.""" - runtime_dir.mkdir(parents=True, exist_ok=True) - framework_dir = Path(__file__).resolve().parent - - for script_name in ("coworker.py", "utils.py"): - script_path = framework_dir / script_name - if script_path.is_file(): - shutil.copy2(script_path, runtime_dir / script_name) - - -def copy_runtime_assets( - package_dir: Path, output: Path, manifest: dict[str, Any] -) -> None: - """Copy configured package assets and bundle framework runtime worker scripts.""" - for item in manifest.get("copy", []): - rel = safe_relative(item, "copy path") - source = rooted(package_dir, str(rel), "copy path") - copy_asset(source, output / rel) - - bundle_runtime_scripts(output / "runtime") - - -def runtime_copy_mappings( - package_name: str, copy_items: Sequence[str] -) -> list[dict[str, str]]: - """Build list of runtime and schema copy mappings for an installed package.""" - mappings = [ - {"source": "runtime", "destination": f".coworker/{package_name}/runtime"}, - {"source": "schemas", "destination": f".coworker/{package_name}/schemas"}, - ] - if any(Path(item).parts[:1] == ("scripts",) for item in copy_items): - mappings.append({ - "source": "scripts", - "destination": f".coworker/{package_name}/scripts", - }) - mappings.append({ - "source": "coworker-build.json", - "destination": f".coworker/{package_name}/coworker-build.json", - }) - return mappings - - -def harness_install_specs(entry_name: str) -> dict[str, dict[str, Any]]: - """Return harness-specific copy rules, version command, and configuration settings.""" - return { - "claude-code": { - "version_command": ["claude", "--version"], - "copies": [ - { - "source": f"project/skills/{entry_name}", - "destination": f".claude/skills/{entry_name}", - }, - {"source": "project/agents", "destination": ".claude/agents"}, - ], - "json_merges": [{ - "path": ".claude/settings.local.json", - "value": { - "env": { - "CLAUDE_CODE_MAX_SUBAGENT_SPAWN_DEPTH": "2", - "CLAUDE_CODE_MAX_CONCURRENT_SUBAGENTS": "2", - } - }, - }], - "toml_sets": [], - }, - "codex": { - "version_command": ["codex", "--version"], - "copies": [ - { - "source": f"skills/{entry_name}", - "destination": f".agents/skills/{entry_name}", - }, - {"source": "agents", "destination": ".codex/agents"}, - ], - "json_merges": [], - "toml_sets": [ - { - "path": ".codex/config.toml", - "section": "agents", - "key": "enabled", - "value": True, - }, - { - "path": ".codex/config.toml", - "section": "agents", - "key": "max_concurrent_threads_per_session", - "value": 2, - }, - ], - }, - } - - -def build_install_plan( - manifest: dict[str, Any], - target: dict[str, Any], - harness: str, -) -> dict[str, Any]: - """Generate coworker-install.json installation plan for target harness.""" - name = manifest["name"] - entry = manifest["entrypoint"]["name"] - spec = harness_install_specs(entry)[harness] - - return { - "harness": harness, - "version_range": target["versions"], - "environment_questions": "environment/questions.json", - "environment_schema": manifest["environment"]["schema"], - "environment_path": f".coworker/{name}/environment.json", - "ownership_path": f".coworker/{name}/ownership.json", - "version_command": spec["version_command"], - "copies": ( - spec["copies"] + runtime_copy_mappings(name, manifest.get("copy", [])) - ), - "json_merges": spec["json_merges"], - "toml_sets": spec["toml_sets"], - } - - -def translate_claude( - package_dir: Path, - output: Path, - manifest: dict[str, Any], - target: dict[str, Any], -) -> None: - """Translate package into Claude Code plugin and project layouts.""" - plugin = output / ".claude-plugin" - write_json( - plugin / "plugin.json", - { - "name": manifest["name"], - "displayName": manifest.get("display_name", manifest["name"]), - "version": manifest["version"], - "description": manifest["description"], - "author": {"name": manifest.get("author", "Coworker Framework")}, - }, - ) - - entry_name = manifest["entrypoint"]["name"] - plugin_agent_prefix = f"{manifest['name']}:" - - # Write plugin skill and standalone project skill - skill = output / "skills" / entry_name / "SKILL.md" - skill.parent.mkdir(parents=True, exist_ok=True) - skill.write_text( - skill_document(package_dir, manifest, target, plugin_agent_prefix), - encoding="utf-8", - ) - - project_skill = output / "project" / "skills" / entry_name / "SKILL.md" - project_skill.parent.mkdir(parents=True, exist_ok=True) - project_skill.write_text( - skill_document(package_dir, manifest, target), encoding="utf-8" - ) - - # Write plugin agents and project agents - for agent in manifest["agents"]: - fm = claude_frontmatter(agent) - path = output / "agents" / f"{agent['name']}.md" - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text( - fm - + agent_prompt( - package_dir, manifest, target, agent, plugin_agent_prefix - ), - encoding="utf-8", - ) - - project_agent = output / "project" / "agents" / f"{agent['name']}.md" - project_agent.parent.mkdir(parents=True, exist_ok=True) - project_agent.write_text( - fm + agent_prompt(package_dir, manifest, target, agent), - encoding="utf-8", - ) - - write_json( - output / "coworker-install.json", - build_install_plan(manifest, target, "claude-code"), - ) - - -def translate_codex( - package_dir: Path, - output: Path, - manifest: dict[str, Any], - target: dict[str, Any], -) -> None: - """Translate package into Codex plugin and agents layouts.""" - plugin = output / ".codex-plugin" - write_json( - plugin / "plugin.json", - { - "name": manifest["name"], - "version": manifest["version"], - "description": manifest["description"], - "author": {"name": manifest.get("author", "Coworker Framework")}, - "skills": "./skills/", - "interface": { - "displayName": manifest.get("display_name", manifest["name"]), - "shortDescription": manifest["description"][:80], - "longDescription": manifest["description"], - "developerName": manifest.get("author", "Coworker Framework"), - "category": "Productivity", - "capabilities": ["Interactive", "Write"], - "defaultPrompt": [ - manifest["entrypoint"].get( - "example_prompt", "Run this workflow." - )[:128] - ], - }, - }, - ) - - entry_name = manifest["entrypoint"]["name"] - skill = output / "skills" / entry_name / "SKILL.md" - skill.parent.mkdir(parents=True, exist_ok=True) - skill.write_text( - skill_document(package_dir, manifest, target), encoding="utf-8" - ) - - for agent in manifest["agents"]: - config = output / "agents" / f"{agent['name']}.toml" - config.parent.mkdir(parents=True, exist_ok=True) - prompt = agent_prompt(package_dir, manifest, target, agent).replace( - '"""', "'''" - ) - sandbox = ( - "workspace-write" - if any(tool in agent["tools"] for tool in ("write", "edit")) - else "read-only" - ) - config.write_text( - f'name = "{agent["name"]}"\n' - f'description = "{agent["description"]}"\n' - f'sandbox_mode = "{sandbox}"\n' - f'developer_instructions = """\n{prompt}\n"""\n', - encoding="utf-8", - ) - - write_json( - output / "coworker-install.json", - build_install_plan(manifest, target, "codex"), - ) - - -def translate(package_dir: Path, harness: str, output: Path) -> None: - """Translate a package into a target harness distribution.""" - manifest = verify_package(package_dir) - target = target_for(manifest, harness) - - if output.exists(): - marker = output / "coworker-build.json" - if not marker.is_file() or not load_json(marker).get("generated"): - raise CoworkerError( - f"refusing to replace non-generated output directory: {output}" - ) - shutil.rmtree(output) - - output.mkdir(parents=True) - - translators: dict[ - str, Callable[[Path, Path, dict[str, Any], dict[str, Any]], None] - ] = { - "claude-code": translate_claude, - "codex": translate_codex, - } - translators[harness](package_dir, output, manifest, target) - - copy_runtime_assets(package_dir, output, manifest) - - write_json( - output / "coworker-build.json", - { - "framework_version": FRAMEWORK_VERSION, - "source": manifest["name"], - "source_version": manifest["version"], - "compatibility_target": target["name"], - "harness": harness, - "version_range": target["versions"], - "generated": True, - }, - ) - - -# ===================================================================== -# Environment Prompting & Question Coercion -# ===================================================================== - - -def coerce_answer_value(question: dict[str, Any], raw: Any) -> Any: - """Coerce raw input into the declared question type and validate choices.""" - key = question["name"] - if question.get("type") == "boolean" and isinstance(raw, str): - if raw.lower() not in {"yes", "no", "true", "false", "y", "n"}: - raise CoworkerError(f"{key} must be yes/no") - raw = raw.lower() in {"yes", "true", "y"} - - if question.get("type") == "integer" and isinstance(raw, str): - try: - raw = int(raw) - except ValueError as exc: - raise CoworkerError(f"{key} must be an integer") from exc - - if "choices" in question and raw not in question["choices"]: - raise CoworkerError(f"{key} must be one of {question['choices']}") - - return raw - - -def prompt_question_value( - question: dict[str, Any], - supplied: dict[str, Any], - current: dict[str, Any], - non_interactive: bool, - destination: Path, -) -> Any: - """Resolve the value for an environment question interactively or from answers.""" - key = question["name"] - if question.get("value_from") == "destination": - return str(destination.resolve()) - - if key in supplied: - return coerce_answer_value(question, supplied[key]) - - fallback = current.get(key, question.get("default")) - - if non_interactive: - if fallback is None: - raise CoworkerError(f"missing required answer: {key}") - return coerce_answer_value(question, fallback) - - choices = ( - f" ({' | '.join(map(str, question['choices']))})" - if "choices" in question - else "" - ) - suffix = f" [{fallback}]" if fallback is not None else "" - entered = input(f"{question['prompt']}{choices}{suffix}: ").strip() - - if not entered and fallback is None: - raise CoworkerError(f"missing required answer: {key}") - - return coerce_answer_value(question, entered if entered else fallback) - - -# Backward-compatible function alias -prompt_value = prompt_question_value - - -def validate_question_constraints( - questions: list[dict[str, Any]], answers: dict[str, Any] -) -> None: - """Validate filesystem and relational constraints on environment answers.""" - for question in questions: - value = answers[question["name"]] - if question.get("relative_path") and Path(str(value)).is_absolute(): - raise CoworkerError(f"{question['name']} must be relative") - - if "must_exist" in question: - candidate = Path(str(value)) - if "relative_to" in question: - candidate = Path(str(answers[question["relative_to"]])) / candidate - expected = question["must_exist"] - if expected == "file" and not candidate.is_file(): - raise CoworkerError( - f"{question['name']} does not identify a file: {candidate}" - ) - if expected == "directory" and not candidate.is_dir(): - raise CoworkerError( - f"{question['name']} does not identify a directory: {candidate}" - ) - - -# ===================================================================== -# Harness Detection & File Expansion -# ===================================================================== - - -def check_harness(plan: dict[str, Any]) -> str: - """Execute harness version command and assert compatibility against plan version range.""" - try: - result = subprocess.run( - plan["version_command"], - text=True, - capture_output=True, - check=True, - ) - except (OSError, subprocess.CalledProcessError) as exc: - raise CoworkerError(f"cannot detect {plan['harness']}: {exc}") from exc - - output = (result.stdout or result.stderr).strip() - if not version_satisfies(SemVer.parse(output), plan["version_range"]): - raise CoworkerError( - f"installed {plan['harness']} version {output!r} is outside" - f" {plan['version_range']}" - ) - return output - - -def expand_copies( - distribution: Path, - destination: Path, - plan: dict[str, Any], -) -> list[tuple[Path, Path]]: - """Expand copy specifications into explicit (source_file, dest_file) pairs.""" - expanded: list[tuple[Path, Path]] = [] - seen: set[Path] = set() - - for mapping in plan["copies"]: - source_rel = safe_relative(mapping["source"], "copy source") - destination_rel = safe_relative(mapping["destination"], "copy destination") - - source = ( - distribution - if str(source_rel) == "." - else rooted(distribution, str(source_rel), "copy source") - ) - target = rooted(destination, str(destination_rel), "copy destination") - - if source.is_file(): - pairs = [(source, target)] - elif source.is_dir(): - pairs = [ - (item, target / item.relative_to(source)) - for item in source.rglob("*") - if item.is_file() - ] - else: - raise CoworkerError(f"install source does not exist: {source}") - - for _, installed in pairs: - if installed in seen: - raise CoworkerError( - f"install plan writes the same file twice: {installed}" - ) - seen.add(installed) - - expanded.extend(pairs) - - return expanded - - -# ===================================================================== -# Ownership Management & Rollback -# ===================================================================== - - -def clean_empty_directories(root: Path) -> None: - """Remove empty directories under root in bottom-up order.""" - for directory in sorted( - (p for p in root.rglob("*") if p.is_dir()), reverse=True - ): - try: - directory.rmdir() - except OSError: - pass - - -def check_action_modification( - destination: Path, action: dict[str, Any] -) -> str | None: - """Check if an ownership action target has been modified by the user.""" - path = rooted(destination, action["path"], "owned path") - kind = action["kind"] - - if kind == "file": - if path.is_file() and compute_digest(path) != action["sha256"]: - return action["path"] - elif kind == "json_value" and path.is_file(): - exists, value = JsonConfigManager.get_at_path( - load_json(path), action["keys"] - ) - if exists and value != action["value"]: - return f"{action['path']}:{'/'.join(action['keys'])}" - elif kind == "json_container" and path.is_file(): - exists, value = JsonConfigManager.get_at_path( - load_json(path), action["keys"] - ) - if exists and not isinstance(value, dict): - return f"{action['path']}:{'/'.join(action['keys'])}" - elif kind == "toml_value" and path.is_file(): - lines = path.read_text(encoding="utf-8").splitlines() - _, value = TomlConfigManager.find_key( - lines, action["section"], action["key"] - ) - if value is not None and value != action["value"]: - return f"{action['path']}:[{action['section']}]/{action['key']}" - - return None - - -def detect_owned_modifications( - destination: Path, ownership: dict[str, Any] -) -> list[str]: - """Detect any files or configuration entries modified since installation.""" - modified: list[str] = [] - for action in ownership.get("actions", []): - mod = check_action_modification(destination, action) - if mod: - modified.append(mod) - return modified - - -def rollback_json_action( - path: Path, - keys: Sequence[str], - expected_val: Any, - is_container: bool = False, -) -> None: - """Roll back an added JSON key or container from a config document.""" - if not path.is_file(): - return - document = load_json(path) - exists, value = JsonConfigManager.get_at_path(document, keys) - if not exists: - return - if (is_container and not value) or ( - not is_container and value == expected_val - ): - parent: Any = document - for key in keys[:-1]: - if isinstance(parent, dict) and key in parent: - parent = parent[key] - else: - parent = None - break - if isinstance(parent, dict) and keys[-1] in parent: - del parent[keys[-1]] - if document: - write_json(path, document) - else: - path.unlink(missing_ok=True) - - -def rollback_toml_action( - path: Path, - section: str, - key: str, - expected_val: Any, - section_created: bool = False, -) -> None: - """Roll back an added TOML key and optional section from a config file.""" - if not path.is_file(): - return - lines = path.read_text(encoding="utf-8").splitlines() - index, value = TomlConfigManager.find_key(lines, section, key) - if index is not None and value == expected_val: - del lines[index] - if section_created: - section_index = next( - (i for i, line in enumerate(lines) if line.strip() == f"[{section}]"), - None, - ) - if section_index is not None: - next_section = next( - ( - i - for i in range(section_index + 1, len(lines)) - if re.match(r"^\s*\[[^]]+]\s*$", lines[i]) - ), - len(lines), - ) - if not any( - line.strip() and not line.lstrip().startswith("#") - for line in lines[section_index + 1 : next_section] - ): - del lines[section_index:next_section] - - content = "\n".join(lines).strip() - if content: - path.write_text("\n".join(lines) + "\n", encoding="utf-8") - else: - path.unlink(missing_ok=True) - - -def rollback_action(destination: Path, action: dict[str, Any]) -> None: - """Revert a single recorded installation action if unchanged.""" - path = rooted(destination, action["path"], "owned path") - kind = action["kind"] - - if kind == "file": - if path.is_file() and compute_digest(path) == action["sha256"]: - path.unlink() - elif kind == "json_value": - rollback_json_action( - path, action["keys"], action["value"], is_container=False - ) - elif kind == "json_container": - rollback_json_action(path, action["keys"], {}, is_container=True) - elif kind == "toml_value": - rollback_toml_action( - path, - action["section"], - action["key"], - action["value"], - action.get("section_created", False), - ) - - -def remove_owned(destination: Path, ownership: dict[str, Any]) -> list[str]: - """Remove files and rollback config settings tracked in ownership record.""" - modified = detect_owned_modifications(destination, ownership) - - for action in reversed(ownership.get("actions", [])): - rollback_action(destination, action) - - clean_empty_directories(destination) - return modified - - -# ===================================================================== -# Install & Uninstall Commands -# ===================================================================== - - -def validate_install_preconditions( - destination: Path, - previous_ownership: dict[str, Any] | None, - upgrade: bool, -) -> None: - """Validate destination state and upgrade requirements before installation.""" - if destination == Path(destination.anchor): - raise CoworkerError("refusing to install into a filesystem root") - - if previous_ownership and not upgrade: - raise CoworkerError( - f"installation already exists in {destination}; use --upgrade" - ) - if upgrade and not previous_ownership: - raise CoworkerError(f"cannot upgrade missing installation in {destination}") - if previous_ownership: - modified = detect_owned_modifications(destination, previous_ownership) - if modified: - raise CoworkerError( - "refusing upgrade because managed content changed:\n" - + "\n".join(modified) - ) - - -def check_file_collisions( - copies: Sequence[tuple[Path, Path]], - destination: Path, - previous_ownership: dict[str, Any] | None, -) -> None: - """Ensure newly copied files do not overwrite unmanaged user files.""" - old_files = { - action["path"] - for action in (previous_ownership or {}).get("actions", []) - if action["kind"] == "file" - } - collisions = [ - str(target) - for _, target in copies - if target.exists() - and str(target.relative_to(destination)) not in old_files - ] - if collisions: - raise CoworkerError( - "refusing to overwrite existing files:\n" + "\n".join(collisions) - ) - - -def prevalidate_configuration_mutations( - destination: Path, plan: dict[str, Any] -) -> None: - """Verify that JSON merges and TOML settings can be applied without conflict.""" - for merge in plan.get("json_merges", []): - path = rooted(destination, merge["path"], "JSON config path") - document = load_json(path) if path.exists() else {} - JsonConfigManager.merge(document, merge["value"], [], []) - - for setting in plan.get("toml_sets", []): - path = rooted(destination, setting["path"], "TOML config path") - lines = ( - path.read_text(encoding="utf-8").splitlines() if path.exists() else [] - ) - _, existing = TomlConfigManager.find_key( - lines, setting["section"], setting["key"] - ) - if existing is not None and existing != TomlConfigManager.render_value( - setting["value"] - ): - raise CoworkerError( - f"refusing to replace existing TOML setting: [{setting['section']}]" - f" {setting['key']}" - ) - - -def collect_environment_answers( - distribution: Path, - destination: Path, - plan: dict[str, Any], - answers_path: Path | None, - previous_environment: dict[str, Any], - non_interactive: bool, -) -> dict[str, Any]: - """Prompt, validate, and return environment configuration answers.""" - spec = load_json( - rooted( - distribution, plan["environment_questions"], "environment questions" - ) - ) - supplied = load_json(answers_path) if answers_path else {} - unknown_answers = set(supplied) - { - question["name"] for question in spec["questions"] - } - if unknown_answers: - raise CoworkerError( - f"unknown answer keys: {', '.join(sorted(unknown_answers))}" - ) - - current = { - key: value - for key, value in previous_environment.items() - if key != "revision" - } - answers = { - q["name"]: prompt_question_value( - q, supplied, current, non_interactive, destination - ) - for q in spec["questions"] - } - validate_question_constraints(spec["questions"], answers) - - env_schema = load_json( - rooted(distribution, plan["environment_schema"], "environment schema") - ) - errors = validate_instance(answers, env_schema) - if errors: - raise CoworkerError("invalid environment:\n" + "\n".join(errors)) - - return answers - - -def install( - distribution: Path, - destination: Path, - answers_path: Path | None, - non_interactive: bool, - upgrade: bool = False, - skip_harness_check: bool = False, -) -> None: - """Install or upgrade a translated distribution package into a workspace.""" - build = load_json(distribution / "coworker-build.json") - plan = load_json(distribution / "coworker-install.json") - - if ( - build["harness"] != plan["harness"] - or build["version_range"] != plan["version_range"] - ): - raise CoworkerError("build metadata and install plan disagree") - - if not skip_harness_check: - check_harness(plan) - - destination = destination.resolve() - ownership_path = rooted(destination, plan["ownership_path"], "ownership path") - environment_path = rooted( - destination, plan["environment_path"], "environment path" - ) - - previous_ownership = ( - load_json(ownership_path) if ownership_path.exists() else None - ) - previous_environment = ( - load_json(environment_path) if environment_path.exists() else {} - ) - - validate_install_preconditions(destination, previous_ownership, upgrade) - answers = collect_environment_answers( - distribution, - destination, - plan, - answers_path, - previous_environment, - non_interactive, - ) - - copies = expand_copies(distribution, destination, plan) - check_file_collisions(copies, destination, previous_ownership) - prevalidate_configuration_mutations(destination, plan) - - # Rollback prior installation if upgrading - if previous_ownership: - remove_owned(destination, previous_ownership) - if ownership_path.exists(): - ownership_path.unlink() - - # Perform installation - destination.mkdir(parents=True, exist_ok=True) - actions: list[dict[str, Any]] = [] - - # Copy files - for source, target in copies: - target.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source, target) - actions.append({ - "kind": "file", - "path": str(target.relative_to(destination)), - "sha256": compute_digest(target), - }) - - # Apply JSON merges - for merge in plan.get("json_merges", []): - path = rooted(destination, merge["path"], "JSON config path") - document = load_json(path) if path.exists() else {} - records: list[dict[str, Any]] = [] - JsonConfigManager.merge(document, merge["value"], [], records) - write_json(path, document) - actions.extend({**record, "path": merge["path"]} for record in records) - - # Apply TOML sets - for setting in plan.get("toml_sets", []): - path = rooted(destination, setting["path"], "TOML config path") - record = TomlConfigManager.apply_set( - path, setting["section"], setting["key"], setting["value"] - ) - if record: - actions.append({**record, "path": setting["path"]}) - - # Update environment revision - current = {k: v for k, v in previous_environment.items() if k != "revision"} - previous_revision = previous_environment.get("revision", 0) - revision = ( - previous_revision + 1 if current != answers else max(previous_revision, 1) - ) - write_json(environment_path, {"revision": revision, **answers}) - actions.append({ - "kind": "file", - "path": str(environment_path.relative_to(destination)), - "sha256": compute_digest(environment_path), - }) - - # Record ownership - ownership_path.parent.mkdir(parents=True, exist_ok=True) - write_json( - ownership_path, - { - "framework_version": FRAMEWORK_VERSION, - "package": build["source"], - "version": build["source_version"], - "harness": build["harness"], - "compatibility_target": build["compatibility_target"], - "actions": actions, - }, - ) - - -def uninstall(destination: Path, package: str) -> list[str]: - """Uninstall an installed package from destination directory, preserving user-modified files.""" - destination = destination.resolve() - validate_package_name(package) - - ownership_path = rooted( - destination, f".coworker/{package}/ownership.json", "ownership path" - ) - ownership = load_json(ownership_path) - modified = remove_owned(destination, ownership) - ownership_path.unlink() - - clean_empty_directories(destination) - return modified - - -# ===================================================================== -# Runtime Commands: Runs & Artifacts -# ===================================================================== - - -def generate_run_id() -> str: - """Generate a collision-resistant timestamped run identifier.""" - timestamp = datetime.datetime.now(datetime.timezone.utc).strftime( - "%Y%m%dT%H%M%SZ" - ) - return f"{timestamp}-{uuid.uuid4().hex[:12]}" - - -def get_installed_package_root(workspace: Path, package: str) -> Path: - """Validate package and return its resolved installation root in the workspace.""" - validate_package_name(package) - package_root = rooted( - workspace.resolve(), f".coworker/{package}", "installed package path" - ) - environment_path = rooted( - package_root, "environment.json", "environment path" - ) - if not environment_path.is_file(): - raise CoworkerError( - f"package {package} is not installed in {workspace.resolve()}" - ) - return package_root - - -def get_installed_run_context( - workspace: Path, package: str, run_id: str -) -> tuple[Path, Path, dict[str, Any]]: - """Retrieve and validate the package root, run root directory, and run metadata document.""" - validate_run_id(run_id) - package_root = get_installed_package_root(workspace, package) - run_root = rooted(package_root, f"runs/{run_id}", "run path") - run_document = load_json(run_root / "run.json") - - if ( - run_document.get("run_id") != run_id - or run_document.get("package") != package - ): - raise CoworkerError( - "run metadata does not match the requested package and run ID" - ) - - return package_root, run_root, run_document - - -def start_run(workspace: Path, package: str) -> dict[str, Any]: - """Create a collision-resistant run namespace inside an installed package.""" - package_root = get_installed_package_root(workspace, package) - environment = load_json(package_root / "environment.json") - runs_root = rooted(package_root, "runs", "runs path") - runs_root.mkdir(parents=True, exist_ok=True) - - for _ in range(10): - run_id = generate_run_id() - run_root = rooted(runs_root, run_id, "run path") - try: - run_root.mkdir() - except FileExistsError: - continue - - (run_root / "artifacts").mkdir() - document = { - "created_at": ( - datetime.datetime.now(datetime.timezone.utc) - .isoformat() - .replace("+00:00", "Z") - ), - "environment_revision": environment.get("revision", 1), - "framework_version": FRAMEWORK_VERSION, - "package": package, - "run_id": run_id, - } - write_json(run_root / "run.json", document) - return document - - raise CoworkerError("could not allocate a unique run ID") - - -def describe_artifact( - workspace: Path, - package: str, - run_id: str, - file: str, - schema: str, - media_type: str, -) -> dict[str, Any]: - """Describe an artifact belonging to an active or completed package run.""" - if not media_type.strip(): - raise CoworkerError("media type must not be empty") - - package_root, run_root, run_document = get_installed_run_context( - workspace, package, run_id - ) - - artifact_root = rooted(run_root, "artifacts", "artifact root") - artifact_file = rooted(artifact_root, file, "artifact file") - if not artifact_file.is_file(): - raise CoworkerError(f"artifact does not identify a file: {artifact_file}") - - schema_path = rooted(package_root, schema, "artifact schema") - if not schema_path.is_file(): - raise CoworkerError( - f"artifact schema does not identify a file: {schema_path}" - ) - - relative = artifact_file.relative_to(artifact_root).as_posix() - return { - "environment_revision": run_document["environment_revision"], - "media_type": media_type, - "run_id": run_id, - "schema": schema, - "sha256": compute_digest(artifact_file), - "uri": f"workspace://{package}/runs/{run_id}/artifacts/{relative}", - } - - -# ===================================================================== -# CLI Entrypoint -# ===================================================================== - - -def build_parser() -> argparse.ArgumentParser: - """Construct command-line argument parser for the Coworker CLI.""" - parser = argparse.ArgumentParser( - prog="coworker", - description=( - "Domain-neutral canonical-package translator, installer, and" - " validator." - ), - ) - sub = parser.add_subparsers(dest="command", required=True) - - # verify - p_verify = sub.add_parser( - "verify", help="Verify package structure and schemas" - ) - p_verify.add_argument("package", type=Path, help="Path to package directory") - - # translate - p_translate = sub.add_parser( - "translate", help="Translate package to target harness" - ) - p_translate.add_argument( - "package", type=Path, help="Path to package directory" - ) - p_translate.add_argument( - "--harness", - choices=["claude-code", "codex"], - required=True, - help="Target harness", - ) - p_translate.add_argument( - "--output", type=Path, required=True, help="Output directory" - ) - - # install - p_install = sub.add_parser( - "install", help="Install translated package into workspace" - ) - p_install.add_argument( - "distribution", - type=Path, - help="Path to translated distribution directory", - ) - p_install.add_argument( - "--destination", - type=Path, - required=True, - help="Workspace destination path", - ) - p_install.add_argument( - "--answers", type=Path, help="Path to JSON file with pre-supplied answers" - ) - p_install.add_argument( - "--non-interactive", - action="store_true", - help="Do not prompt for missing answers", - ) - p_install.add_argument( - "--upgrade", action="store_true", help="Upgrade existing installation" - ) - p_install.add_argument( - "--skip-harness-check", action="store_true", help=argparse.SUPPRESS - ) - - # uninstall - p_uninstall = sub.add_parser( - "uninstall", help="Uninstall package from workspace" - ) - p_uninstall.add_argument( - "destination", type=Path, help="Workspace destination path" - ) - p_uninstall.add_argument( - "--package", required=True, help="Package name to uninstall" - ) - - # validate - p_validate = sub.add_parser( - "validate", help="Validate a JSON instance against a schema" - ) - p_validate.add_argument( - "--schema", type=Path, required=True, help="Path to schema JSON file" - ) - p_validate.add_argument( - "instance", type=Path, help="Path to JSON instance file to validate" - ) - - # start-run - p_start = sub.add_parser( - "start-run", help="Initialize a new run namespace for a package" - ) - p_start.add_argument( - "--workspace", type=Path, required=True, help="Workspace root directory" - ) - p_start.add_argument( - "--package", required=True, help="Installed package name" - ) - - # describe-artifact - p_artifact = sub.add_parser( - "describe-artifact", help="Generate an artifact provenance descriptor" - ) - p_artifact.add_argument( - "--workspace", type=Path, required=True, help="Workspace root directory" - ) - p_artifact.add_argument( - "--package", required=True, help="Installed package name" - ) - p_artifact.add_argument("--run-id", required=True, help="Run ID") - p_artifact.add_argument( - "--file", - required=True, - help="Artifact file path relative to run artifacts/", - ) - p_artifact.add_argument( - "--schema", required=True, help="Schema path relative to package root" - ) - p_artifact.add_argument( - "--media-type", required=True, help="Media type of the artifact" - ) - - return parser - - -def main(argv: list[str] | None = None) -> int: - """Main CLI execution dispatch function.""" - parser = build_parser() - args = parser.parse_args(argv) - - try: - if args.command == "verify": - verify_package(args.package) - - elif args.command == "translate": - translate(args.package, args.harness, args.output) - - elif args.command == "install": - install( - distribution=args.distribution, - destination=args.destination, - answers_path=args.answers, - non_interactive=args.non_interactive, - upgrade=args.upgrade, - skip_harness_check=args.skip_harness_check, - ) - - elif args.command == "uninstall": - modified = uninstall(args.destination, args.package) - if modified: - print( - "Preserved modified managed content:\n" + "\n".join(modified), - file=sys.stderr, - ) - - elif args.command == "validate": - errors = validate_instance( - load_json(args.instance), load_json(args.schema) - ) - if errors: - print("\n".join(errors), file=sys.stderr) - return 2 - - elif args.command == "start-run": - run_doc = start_run(args.workspace, args.package) - print(format_json(run_doc)) - - elif args.command == "describe-artifact": - descriptor = describe_artifact( - workspace=args.workspace, - package=args.package, - run_id=args.run_id, - file=args.file, - schema=args.schema, - media_type=args.media_type, - ) - print(format_json(descriptor)) - - return 0 - - except CoworkerError as exc: - print(f"coworker: {exc}", file=sys.stderr) - return 1 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/nexus/framework/installer.py b/nexus/framework/installer.py new file mode 100644 index 0000000..6d21353 --- /dev/null +++ b/nexus/framework/installer.py @@ -0,0 +1,705 @@ +#!/usr/bin/env python3 +"""Installer module for Coworker framework. + +Handles installing, updating, and cleanly uninstalling translated distributions +in destination workspaces with atomic ownership tracking and rollback support. +""" + +from __future__ import annotations + +import argparse +import pathlib +import re +import shutil +import subprocess +import sys +from typing import Any, Sequence + +try: + from accelerator_agents.tpu_nexus.framework import utils +except ImportError: + try: + from . import utils + except ImportError: + import utils + +Path = pathlib.Path + +CoworkerError = utils.CoworkerError +FRAMEWORK_VERSION = utils.FRAMEWORK_VERSION +JsonConfigManager = utils.JsonConfigManager +SemVer = utils.SemVer +TomlConfigManager = utils.TomlConfigManager +compute_digest = utils.compute_digest +load_json = utils.load_json +rooted = utils.rooted +safe_relative = utils.safe_relative +validate_instance = utils.validate_instance +validate_package_name = utils.validate_package_name +version_satisfies = utils.version_satisfies +write_json = utils.write_json + + +# ===================================================================== +# Environment Prompting & Question Coercion +# ===================================================================== + + +def coerce_answer_value(question: dict[str, Any], raw: Any) -> Any: + """Coerce raw input into the declared question type and validate choices.""" + key = question["name"] + if question.get("type") == "boolean" and isinstance(raw, str): + if raw.lower() not in {"yes", "no", "true", "false", "y", "n"}: + raise CoworkerError(f"{key} must be yes/no") + raw = raw.lower() in {"yes", "true", "y"} + + if question.get("type") == "integer" and isinstance(raw, str): + try: + raw = int(raw) + except ValueError as exc: + raise CoworkerError(f"{key} must be an integer") from exc + + if "choices" in question and raw not in question["choices"]: + raise CoworkerError(f"{key} must be one of {question['choices']}") + + return raw + + +def prompt_question_value( + question: dict[str, Any], + supplied: dict[str, Any], + current: dict[str, Any], + non_interactive: bool, + destination: Path, +) -> Any: + """Resolve the value for an environment question interactively or from answers.""" + key = question["name"] + if question.get("value_from") == "destination": + return str(destination.resolve()) + + if key in supplied: + return coerce_answer_value(question, supplied[key]) + + fallback = current.get(key, question.get("default")) + + if non_interactive: + if fallback is None: + raise CoworkerError(f"missing required answer: {key}") + return coerce_answer_value(question, fallback) + + choices = ( + f" ({' | '.join(map(str, question['choices']))})" + if "choices" in question + else "" + ) + suffix = f" [{fallback}]" if fallback is not None else "" + entered = input(f"{question['prompt']}{choices}{suffix}: ").strip() + + if not entered and fallback is None: + raise CoworkerError(f"missing required answer: {key}") + + return coerce_answer_value(question, entered if entered else fallback) + + +prompt_value = prompt_question_value + + +def validate_question_constraints( + questions: list[dict[str, Any]], answers: dict[str, Any] +) -> None: + """Validate filesystem and relational constraints on environment answers.""" + for question in questions: + value = answers[question["name"]] + if question.get("relative_path") and Path(str(value)).is_absolute(): + raise CoworkerError(f"{question['name']} must be relative") + + if "must_exist" in question: + candidate = Path(str(value)) + if "relative_to" in question: + candidate = Path(str(answers[question["relative_to"]])) / candidate + expected = question["must_exist"] + if expected == "file" and not candidate.is_file(): + raise CoworkerError( + f"{question['name']} does not identify a file: {candidate}" + ) + if expected == "directory" and not candidate.is_dir(): + raise CoworkerError( + f"{question['name']} does not identify a directory: {candidate}" + ) + + +# ===================================================================== +# Harness Detection & File Expansion +# ===================================================================== + + +def check_harness(plan: dict[str, Any]) -> str: + """Execute harness version command and assert compatibility against plan version range.""" + try: + result = subprocess.run( + plan["version_command"], + text=True, + capture_output=True, + check=True, + ) + except (OSError, subprocess.CalledProcessError) as exc: + raise CoworkerError(f"cannot detect {plan['harness']}: {exc}") from exc + + output = (result.stdout or result.stderr).strip() + if not version_satisfies(SemVer.parse(output), plan["version_range"]): + raise CoworkerError( + f"installed {plan['harness']} version {output!r} is outside" + f" {plan['version_range']}" + ) + return output + + +def expand_copies( + distribution: Path, + destination: Path, + plan: dict[str, Any], +) -> list[tuple[Path, Path]]: + """Expand copy specifications into explicit (source_file, dest_file) pairs.""" + expanded: list[tuple[Path, Path]] = [] + seen: set[Path] = set() + + for mapping in plan["copies"]: + source_rel = safe_relative(mapping["source"], "copy source") + destination_rel = safe_relative(mapping["destination"], "copy destination") + + source = ( + distribution + if str(source_rel) == "." + else rooted(distribution, str(source_rel), "copy source") + ) + target = rooted(destination, str(destination_rel), "copy destination") + + if source.is_file(): + pairs = [(source, target)] + elif source.is_dir(): + pairs = [ + (item, target / item.relative_to(source)) + for item in source.rglob("*") + if item.is_file() + ] + else: + raise CoworkerError(f"install source does not exist: {source}") + + for _, installed in pairs: + if installed in seen: + raise CoworkerError( + f"install plan writes the same file twice: {installed}" + ) + seen.add(installed) + + expanded.extend(pairs) + + return expanded + + +# ===================================================================== +# Ownership Management & Rollback +# ===================================================================== + + +def clean_empty_directories(root: Path) -> None: + """Remove empty directories under root in bottom-up order.""" + for directory in sorted( + (p for p in root.rglob("*") if p.is_dir()), reverse=True + ): + try: + directory.rmdir() + except OSError: + pass + + +def check_action_modification( + destination: Path, action: dict[str, Any] +) -> str | None: + """Check if an ownership action target has been modified by the user.""" + path = rooted(destination, action["path"], "owned path") + kind = action["kind"] + + if kind == "file": + if path.is_file() and compute_digest(path) != action["sha256"]: + return action["path"] + elif kind == "json_value" and path.is_file(): + exists, value = JsonConfigManager.get_at_path( + load_json(path), action["keys"] + ) + if exists and value != action["value"]: + return f"{action['path']}:{'/'.join(action['keys'])}" + elif kind == "json_container" and path.is_file(): + exists, value = JsonConfigManager.get_at_path( + load_json(path), action["keys"] + ) + if exists and not isinstance(value, dict): + return f"{action['path']}:{'/'.join(action['keys'])}" + elif kind == "toml_value" and path.is_file(): + lines = path.read_text(encoding="utf-8").splitlines() + _, value = TomlConfigManager.find_key( + lines, action["section"], action["key"] + ) + if value is not None and value != action["value"]: + return f"{action['path']}:[{action['section']}]/{action['key']}" + + return None + + +def detect_owned_modifications( + destination: Path, ownership: dict[str, Any] +) -> list[str]: + """Detect any files or configuration entries modified since installation.""" + modified: list[str] = [] + for action in ownership.get("actions", []): + mod = check_action_modification(destination, action) + if mod: + modified.append(mod) + return modified + + +def rollback_json_action( + path: Path, + keys: Sequence[str], + expected_val: Any, + is_container: bool = False, +) -> None: + """Roll back an added JSON key or container from a config document.""" + if not path.is_file(): + return + document = load_json(path) + exists, value = JsonConfigManager.get_at_path(document, keys) + if not exists: + return + if (is_container and not value) or ( + not is_container and value == expected_val + ): + parent: Any = document + for key in keys[:-1]: + if isinstance(parent, dict) and key in parent: + parent = parent[key] + else: + parent = None + break + if isinstance(parent, dict) and keys[-1] in parent: + del parent[keys[-1]] + if document: + write_json(path, document) + else: + path.unlink(missing_ok=True) + + +def rollback_toml_action( + path: Path, + section: str, + key: str, + expected_val: Any, + section_created: bool = False, +) -> None: + """Roll back an added TOML key and optional section from a config file.""" + if not path.is_file(): + return + lines = path.read_text(encoding="utf-8").splitlines() + index, value = TomlConfigManager.find_key(lines, section, key) + if index is not None and value == expected_val: + del lines[index] + if section_created: + section_index = next( + (i for i, line in enumerate(lines) if line.strip() == f"[{section}]"), + None, + ) + if section_index is not None: + next_section = next( + ( + i + for i in range(section_index + 1, len(lines)) + if re.match(r"^\s*\[[^]]+]\s*$", lines[i]) + ), + len(lines), + ) + if not any( + line.strip() and not line.lstrip().startswith("#") + for line in lines[section_index + 1 : next_section] + ): + del lines[section_index:next_section] + + content = "\n".join(lines).strip() + if content: + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + else: + path.unlink(missing_ok=True) + + +def rollback_action(destination: Path, action: dict[str, Any]) -> None: + """Revert a single recorded installation action if unchanged.""" + path = rooted(destination, action["path"], "owned path") + kind = action["kind"] + + if kind == "file": + if path.is_file() and compute_digest(path) == action["sha256"]: + path.unlink() + elif kind == "json_value": + rollback_json_action( + path, action["keys"], action["value"], is_container=False + ) + elif kind == "json_container": + rollback_json_action(path, action["keys"], {}, is_container=True) + elif kind == "toml_value": + rollback_toml_action( + path, + action["section"], + action["key"], + action["value"], + action.get("section_created", False), + ) + + +def remove_owned(destination: Path, ownership: dict[str, Any]) -> list[str]: + """Remove files and rollback config settings tracked in ownership record.""" + modified = detect_owned_modifications(destination, ownership) + + for action in reversed(ownership.get("actions", [])): + rollback_action(destination, action) + + clean_empty_directories(destination) + return modified + + +# ===================================================================== +# Install & Uninstall Operations +# ===================================================================== + + +def validate_install_preconditions( + destination: Path, + previous_ownership: dict[str, Any] | None, + upgrade: bool, +) -> None: + """Validate destination state and upgrade requirements before installation.""" + if destination == Path(destination.anchor): + raise CoworkerError("refusing to install into a filesystem root") + + if previous_ownership and not upgrade: + raise CoworkerError( + f"installation already exists in {destination}; use --upgrade" + ) + if upgrade and not previous_ownership: + raise CoworkerError(f"cannot upgrade missing installation in {destination}") + if previous_ownership: + modified = detect_owned_modifications(destination, previous_ownership) + if modified: + raise CoworkerError( + "refusing upgrade because managed content changed:\n" + + "\n".join(modified) + ) + + +def check_file_collisions( + copies: Sequence[tuple[Path, Path]], + destination: Path, + previous_ownership: dict[str, Any] | None, +) -> None: + """Ensure newly copied files do not overwrite unmanaged user files.""" + old_files = { + action["path"] + for action in (previous_ownership or {}).get("actions", []) + if action["kind"] == "file" + } + collisions = [ + str(target) + for _, target in copies + if target.exists() + and str(target.relative_to(destination)) not in old_files + ] + if collisions: + raise CoworkerError( + "refusing to overwrite existing files:\n" + "\n".join(collisions) + ) + + +def prevalidate_configuration_mutations( + destination: Path, plan: dict[str, Any] +) -> None: + """Verify that JSON merges and TOML settings can be applied without conflict.""" + for merge in plan.get("json_merges", []): + path = rooted(destination, merge["path"], "JSON config path") + document = load_json(path) if path.exists() else {} + JsonConfigManager.merge(document, merge["value"], [], []) + + for setting in plan.get("toml_sets", []): + path = rooted(destination, setting["path"], "TOML config path") + lines = ( + path.read_text(encoding="utf-8").splitlines() if path.exists() else [] + ) + _, existing = TomlConfigManager.find_key( + lines, setting["section"], setting["key"] + ) + if existing is not None and existing != TomlConfigManager.render_value( + setting["value"] + ): + raise CoworkerError( + f"refusing to replace existing TOML setting: [{setting['section']}]" + f" {setting['key']}" + ) + + +def collect_environment_answers( + distribution: Path, + destination: Path, + plan: dict[str, Any], + answers_path: Path | None, + previous_environment: dict[str, Any], + non_interactive: bool, +) -> dict[str, Any]: + """Prompt, validate, and return environment configuration answers.""" + spec = load_json( + rooted( + distribution, plan["environment_questions"], "environment questions" + ) + ) + supplied = load_json(answers_path) if answers_path else {} + unknown_answers = set(supplied) - { + question["name"] for question in spec["questions"] + } + if unknown_answers: + raise CoworkerError( + f"unknown answer keys: {', '.join(sorted(unknown_answers))}" + ) + + current = { + key: value + for key, value in previous_environment.items() + if key != "revision" + } + answers = { + q["name"]: prompt_question_value( + q, supplied, current, non_interactive, destination + ) + for q in spec["questions"] + } + validate_question_constraints(spec["questions"], answers) + + env_schema = load_json( + rooted(distribution, plan["environment_schema"], "environment schema") + ) + errors = validate_instance(answers, env_schema) + if errors: + raise CoworkerError("invalid environment:\n" + "\n".join(errors)) + + return answers + + +def install( + distribution: Path, + destination: Path, + answers_path: Path | None = None, + non_interactive: bool = False, + upgrade: bool = False, + skip_harness_check: bool = False, +) -> None: + """Install or upgrade a translated distribution package into a workspace.""" + build = load_json(distribution / "coworker-build.json") + plan = load_json(distribution / "coworker-install.json") + + if ( + build["harness"] != plan["harness"] + or build["version_range"] != plan["version_range"] + ): + raise CoworkerError("build metadata and install plan disagree") + + if not skip_harness_check: + check_harness(plan) + + destination = destination.resolve() + ownership_path = rooted(destination, plan["ownership_path"], "ownership path") + environment_path = rooted( + destination, plan["environment_path"], "environment path" + ) + + previous_ownership = ( + load_json(ownership_path) if ownership_path.exists() else None + ) + previous_environment = ( + load_json(environment_path) if environment_path.exists() else {} + ) + + validate_install_preconditions(destination, previous_ownership, upgrade) + answers = collect_environment_answers( + distribution, + destination, + plan, + answers_path, + previous_environment, + non_interactive, + ) + + copies = expand_copies(distribution, destination, plan) + check_file_collisions(copies, destination, previous_ownership) + prevalidate_configuration_mutations(destination, plan) + + # Rollback prior installation if upgrading + if previous_ownership: + remove_owned(destination, previous_ownership) + if ownership_path.exists(): + ownership_path.unlink() + + # Perform installation + destination.mkdir(parents=True, exist_ok=True) + actions: list[dict[str, Any]] = [] + + # Copy files + for source, target in copies: + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, target) + actions.append({ + "kind": "file", + "path": str(target.relative_to(destination)), + "sha256": compute_digest(target), + }) + + # Apply JSON merges + for merge in plan.get("json_merges", []): + path = rooted(destination, merge["path"], "JSON config path") + document = load_json(path) if path.exists() else {} + records: list[dict[str, Any]] = [] + JsonConfigManager.merge(document, merge["value"], [], records) + write_json(path, document) + actions.extend({**record, "path": merge["path"]} for record in records) + + # Apply TOML sets + for setting in plan.get("toml_sets", []): + path = rooted(destination, setting["path"], "TOML config path") + record = TomlConfigManager.apply_set( + path, setting["section"], setting["key"], setting["value"] + ) + if record: + actions.append({**record, "path": setting["path"]}) + + # Update environment revision + current = {k: v for k, v in previous_environment.items() if k != "revision"} + previous_revision = previous_environment.get("revision", 0) + revision = ( + previous_revision + 1 if current != answers else max(previous_revision, 1) + ) + write_json(environment_path, {"revision": revision, **answers}) + actions.append({ + "kind": "file", + "path": str(environment_path.relative_to(destination)), + "sha256": compute_digest(environment_path), + }) + + # Record ownership + ownership_path.parent.mkdir(parents=True, exist_ok=True) + write_json( + ownership_path, + { + "framework_version": FRAMEWORK_VERSION, + "package": build["source"], + "version": build["source_version"], + "harness": build["harness"], + "compatibility_target": build["compatibility_target"], + "actions": actions, + }, + ) + + +def uninstall(destination: Path, package: str) -> list[str]: + """Uninstall an installed package from destination directory, preserving user-modified files.""" + destination = destination.resolve() + validate_package_name(package) + + ownership_path = rooted( + destination, f".coworker/{package}/ownership.json", "ownership path" + ) + ownership = load_json(ownership_path) + modified = remove_owned(destination, ownership) + ownership_path.unlink() + + clean_empty_directories(destination) + return modified + + +def build_installer_parser() -> argparse.ArgumentParser: + """Construct parser for standalone installer commands.""" + parser = argparse.ArgumentParser( + prog="coworker-installer", + description="Installation and uninstallation operations for Coworker packages.", + ) + sub = parser.add_subparsers(dest="command", required=True) + + # install + p_install = sub.add_parser( + "install", help="Install translated package into workspace" + ) + p_install.add_argument( + "distribution", + type=Path, + help="Path to translated distribution directory", + ) + p_install.add_argument( + "--destination", + type=Path, + required=True, + help="Workspace destination path", + ) + p_install.add_argument( + "--answers", type=Path, help="Path to JSON file with pre-supplied answers" + ) + p_install.add_argument( + "--non-interactive", + action="store_true", + help="Do not prompt for missing answers", + ) + p_install.add_argument( + "--upgrade", action="store_true", help="Upgrade existing installation" + ) + p_install.add_argument( + "--skip-harness-check", action="store_true", help=argparse.SUPPRESS + ) + + # uninstall + p_uninstall = sub.add_parser( + "uninstall", help="Uninstall package from workspace" + ) + p_uninstall.add_argument( + "destination", type=Path, help="Workspace destination path" + ) + p_uninstall.add_argument( + "--package", required=True, help="Package name to uninstall" + ) + + return parser + + +def main(argv: list[str] | None = None) -> int: + """Main entrypoint for standalone installer operations.""" + parser = build_installer_parser() + args = parser.parse_args(argv) + + try: + if args.command == "install": + install( + distribution=args.distribution, + destination=args.destination, + answers_path=args.answers, + non_interactive=args.non_interactive, + upgrade=args.upgrade, + skip_harness_check=args.skip_harness_check, + ) + + elif args.command == "uninstall": + modified = uninstall(args.destination, args.package) + if modified: + print( + "Preserved modified managed content:\n" + "\n".join(modified), + file=sys.stderr, + ) + + return 0 + except CoworkerError as exc: + print(f"installer: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/nexus/framework/runtime.py b/nexus/framework/runtime.py new file mode 100644 index 0000000..b489d82 --- /dev/null +++ b/nexus/framework/runtime.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +"""Embedded runtime for Coworker agent packages. + +Provides runtime execution support for run namespace allocation, artifact +provenance and descriptor generation, and deterministic payload validation. +""" + +from __future__ import annotations + +import argparse +import datetime +import pathlib +import sys +from typing import Any +import uuid + +try: + from accelerator_agents.tpu_nexus.framework import utils +except ImportError: + try: + from . import utils + except ImportError: + import utils + +Path = pathlib.Path + +CoworkerError = utils.CoworkerError +FRAMEWORK_VERSION = utils.FRAMEWORK_VERSION +compute_digest = utils.compute_digest +format_json = utils.format_json +load_json = utils.load_json +rooted = utils.rooted +validate_instance = utils.validate_instance +validate_package_name = utils.validate_package_name +validate_run_id = utils.validate_run_id +write_json = utils.write_json + + +def generate_run_id() -> str: + """Generate a collision-resistant timestamped run identifier.""" + timestamp = datetime.datetime.now(datetime.timezone.utc).strftime( + "%Y%m%dT%H%M%SZ" + ) + return f"{timestamp}-{uuid.uuid4().hex[:12]}" + + +def get_installed_package_root(workspace: Path, package: str) -> Path: + """Validate package and return its resolved installation root in the workspace.""" + validate_package_name(package) + package_root = rooted( + workspace.resolve(), f".coworker/{package}", "installed package path" + ) + environment_path = rooted( + package_root, "environment.json", "environment path" + ) + if not environment_path.is_file(): + raise CoworkerError( + f"package {package} is not installed in {workspace.resolve()}" + ) + return package_root + + +def get_installed_run_context( + workspace: Path, package: str, run_id: str +) -> tuple[Path, Path, dict[str, Any]]: + """Retrieve and validate the package root, run root directory, and run metadata document.""" + validate_run_id(run_id) + package_root = get_installed_package_root(workspace, package) + run_root = rooted(package_root, f"runs/{run_id}", "run path") + run_document = load_json(run_root / "run.json") + + if ( + run_document.get("run_id") != run_id + or run_document.get("package") != package + ): + raise CoworkerError( + "run metadata does not match the requested package and run ID" + ) + + return package_root, run_root, run_document + + +def start_run(workspace: Path, package: str) -> dict[str, Any]: + """Create a collision-resistant run namespace inside an installed package.""" + package_root = get_installed_package_root(workspace, package) + environment = load_json(package_root / "environment.json") + runs_root = rooted(package_root, "runs", "runs path") + runs_root.mkdir(parents=True, exist_ok=True) + + for _ in range(10): + run_id = generate_run_id() + run_root = rooted(runs_root, run_id, "run path") + try: + run_root.mkdir() + except FileExistsError: + continue + + (run_root / "artifacts").mkdir() + document = { + "created_at": ( + datetime.datetime.now(datetime.timezone.utc) + .isoformat() + .replace("+00:00", "Z") + ), + "environment_revision": environment.get("revision", 1), + "framework_version": FRAMEWORK_VERSION, + "package": package, + "run_id": run_id, + } + write_json(run_root / "run.json", document) + return document + + raise CoworkerError("could not allocate a unique run ID") + + +def describe_artifact( + workspace: Path, + package: str, + run_id: str, + file: str, + schema: str, + media_type: str, +) -> dict[str, Any]: + """Describe an artifact belonging to an active or completed package run.""" + if not media_type.strip(): + raise CoworkerError("media type must not be empty") + + package_root, run_root, run_document = get_installed_run_context( + workspace, package, run_id + ) + + artifact_root = rooted(run_root, "artifacts", "artifact root") + artifact_file = rooted(artifact_root, file, "artifact file") + if not artifact_file.is_file(): + raise CoworkerError(f"artifact does not identify a file: {artifact_file}") + + schema_path = rooted(package_root, schema, "artifact schema") + if not schema_path.is_file(): + raise CoworkerError( + f"artifact schema does not identify a file: {schema_path}" + ) + + relative = artifact_file.relative_to(artifact_root).as_posix() + return { + "environment_revision": run_document["environment_revision"], + "media_type": media_type, + "run_id": run_id, + "schema": schema, + "sha256": compute_digest(artifact_file), + "uri": f"workspace://{package}/runs/{run_id}/artifacts/{relative}", + } + + +def validate_json_file(schema_path: Path, instance_path: Path) -> list[str]: + """Validate a JSON instance against a JSON schema file.""" + instance = load_json(instance_path) + schema = load_json(schema_path) + return validate_instance(instance, schema) + + +def build_runtime_parser() -> argparse.ArgumentParser: + """Construct parser for standalone runtime commands.""" + parser = argparse.ArgumentParser( + prog="coworker-runtime", + description="Embedded runtime execution operations for Coworker packages.", + ) + sub = parser.add_subparsers(dest="command", required=True) + + # start-run + p_start = sub.add_parser( + "start-run", help="Initialize a new run namespace for a package" + ) + p_start.add_argument( + "--workspace", type=Path, required=True, help="Workspace root directory" + ) + p_start.add_argument( + "--package", required=True, help="Installed package name" + ) + + # describe-artifact + p_artifact = sub.add_parser( + "describe-artifact", help="Generate an artifact provenance descriptor" + ) + p_artifact.add_argument( + "--workspace", type=Path, required=True, help="Workspace root directory" + ) + p_artifact.add_argument( + "--package", required=True, help="Installed package name" + ) + p_artifact.add_argument("--run-id", required=True, help="Run ID") + p_artifact.add_argument( + "--file", + required=True, + help="Artifact file path relative to run artifacts/", + ) + p_artifact.add_argument( + "--schema", required=True, help="Schema path relative to package root" + ) + p_artifact.add_argument( + "--media-type", required=True, help="Media type of the artifact" + ) + + # validate + p_validate = sub.add_parser( + "validate", help="Validate a JSON instance against a schema" + ) + p_validate.add_argument( + "--schema", type=Path, required=True, help="Path to schema JSON file" + ) + p_validate.add_argument( + "instance", type=Path, help="Path to JSON instance file to validate" + ) + + return parser + + +def main(argv: list[str] | None = None) -> int: + """Main entrypoint for standalone runtime operations.""" + parser = build_runtime_parser() + args = parser.parse_args(argv) + + try: + if args.command == "start-run": + run_doc = start_run(args.workspace, args.package) + print(format_json(run_doc)) + + elif args.command == "describe-artifact": + descriptor = describe_artifact( + workspace=args.workspace, + package=args.package, + run_id=args.run_id, + file=args.file, + schema=args.schema, + media_type=args.media_type, + ) + print(format_json(descriptor)) + + elif args.command == "validate": + errors = validate_json_file(args.schema, args.instance) + if errors: + print("\n".join(errors), file=sys.stderr) + return 2 + + return 0 + except CoworkerError as exc: + print(f"runtime: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/nexus/framework/tests/test_compiler.py b/nexus/framework/tests/test_compiler.py new file mode 100644 index 0000000..9ae2ecc --- /dev/null +++ b/nexus/framework/tests/test_compiler.py @@ -0,0 +1,179 @@ +"""Unit tests for compiler module (verification and translation).""" + +import json +import pathlib +import shutil +import tempfile +import unittest + +try: + from accelerator_agents.tpu_nexus.framework import compiler, utils +except ImportError: + try: + from .. import compiler, utils + except ImportError: + import compiler + import utils + +Path = pathlib.Path + + +class TestCompiler(unittest.TestCase): + """Unit tests for package verification, delegation graph, and translation.""" + + def setUp(self): + super().setUp() + self.temp_dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + super().tearDown() + + def _create_valid_package(self) -> Path: + pkg_dir = self.temp_dir / "valid-pkg" + pkg_dir.mkdir(parents=True, exist_ok=True) + manifest = { + "name": "valid-pkg", + "version": "1.0.0", + "description": "Valid test package", + "entrypoint": { + "name": "entry", + "instructions": "instructions/entry.md", + "delegates": ["worker"], + }, + "agents": [{ + "name": "worker", + "description": "Worker", + "instructions": "instructions/worker.md", + "accepts": "schemas/in.json", + "produces": "schemas/out.json", + "delegates": [], + "tools": ["read", "write"], + }], + "copy": ["schemas", "environment"], + "environment": { + "questions": "environment/questions.json", + "schema": "schemas/env.json", + }, + "compatibility_targets": [ + { + "name": "claude", + "harness": "claude-code", + "versions": ">=2.0.0", + "capabilities": {}, + }, + { + "name": "codex", + "harness": "codex", + "versions": ">=0.145.0", + "capabilities": {}, + }, + ], + "schemas": ["schemas/in.json", "schemas/out.json", "schemas/env.json"], + } + (pkg_dir / "package.json").write_text( + json.dumps(manifest), encoding="utf-8" + ) + (pkg_dir / "instructions").mkdir(parents=True, exist_ok=True) + (pkg_dir / "instructions" / "entry.md").write_text( + "Call {{agent:worker}}.", encoding="utf-8" + ) + (pkg_dir / "instructions" / "worker.md").write_text( + "Work.", encoding="utf-8" + ) + (pkg_dir / "schemas").mkdir(parents=True, exist_ok=True) + (pkg_dir / "schemas" / "in.json").write_text( + json.dumps({"type": "object"}), encoding="utf-8" + ) + (pkg_dir / "schemas" / "out.json").write_text( + json.dumps({"type": "object"}), encoding="utf-8" + ) + (pkg_dir / "schemas" / "env.json").write_text( + json.dumps({"type": "object"}), encoding="utf-8" + ) + (pkg_dir / "environment").mkdir(parents=True, exist_ok=True) + (pkg_dir / "environment" / "questions.json").write_text( + json.dumps({"questions": []}), encoding="utf-8" + ) + return pkg_dir + + def test_verify_valid_package(self): + pkg_dir = self._create_valid_package() + # Should complete without error + compiler.verify_package(pkg_dir) + + def test_cycle_detection(self): + pkg_dir = self.temp_dir / "cycle-pkg" + pkg_dir.mkdir() + manifest = { + "name": "cycle-pkg", + "version": "1.0.0", + "description": "Cycle", + "entrypoint": { + "name": "entry", + "instructions": "entry.md", + "delegates": ["a"], + }, + "agents": [ + { + "name": "a", + "description": "a", + "instructions": "a.md", + "accepts": "s.json", + "produces": "s.json", + "delegates": ["b"], + "tools": ["read"], + }, + { + "name": "b", + "description": "b", + "instructions": "b.md", + "accepts": "s.json", + "produces": "s.json", + "delegates": ["a"], + "tools": ["read"], + }, + ], + "environment": {"questions": "q.json", "schema": "s.json"}, + "compatibility_targets": [ + {"name": "c", "harness": "claude-code", "versions": ">=1.0.0"} + ], + } + (pkg_dir / "package.json").write_text(json.dumps(manifest), encoding="utf-8") + (pkg_dir / "entry.md").write_text("entry", encoding="utf-8") + (pkg_dir / "a.md").write_text("a", encoding="utf-8") + (pkg_dir / "b.md").write_text("b", encoding="utf-8") + (pkg_dir / "s.json").write_text("{}", encoding="utf-8") + (pkg_dir / "q.json").write_text( + json.dumps({"questions": []}), encoding="utf-8" + ) + + with self.assertRaises(utils.CoworkerError) as ctx: + compiler.load_package(pkg_dir) + self.assertIn("cycle", str(ctx.exception).lower()) + + def test_translate_claude(self): + pkg_dir = self._create_valid_package() + dist_dir = self.temp_dir / "dist_claude" + compiler.translate(pkg_dir, "claude-code", dist_dir) + + self.assertTrue((dist_dir / ".claude-plugin" / "plugin.json").is_file()) + self.assertTrue((dist_dir / "skills" / "entry" / "SKILL.md").is_file()) + self.assertTrue((dist_dir / "agents" / "worker.md").is_file()) + self.assertTrue((dist_dir / "coworker-build.json").is_file()) + self.assertTrue((dist_dir / "coworker-install.json").is_file()) + self.assertTrue((dist_dir / "runtime" / "runtime.py").is_file()) + self.assertTrue((dist_dir / "runtime" / "utils.py").is_file()) + + def test_translate_codex(self): + pkg_dir = self._create_valid_package() + dist_dir = self.temp_dir / "dist_codex" + compiler.translate(pkg_dir, "codex", dist_dir) + + self.assertTrue((dist_dir / ".codex-plugin" / "plugin.json").is_file()) + self.assertTrue((dist_dir / "skills" / "entry" / "SKILL.md").is_file()) + self.assertTrue((dist_dir / "agents" / "worker.toml").is_file()) + + +if __name__ == "__main__": + unittest.main() diff --git a/nexus/framework/tests/test_coworker.py b/nexus/framework/tests/test_coworker.py deleted file mode 100644 index 171e7c7..0000000 --- a/nexus/framework/tests/test_coworker.py +++ /dev/null @@ -1,536 +0,0 @@ -"""Unit tests for the Coworker agent framework.""" - -import json -import pathlib -import shutil -import tempfile -import unittest - -from nexus.framework import coworker as cw - - -Path = pathlib.Path - - -class TestCoworker(unittest.TestCase): - """Unit tests for coworker framework verification, translation, and lifecycle.""" - - def setUp(self): - super().setUp() - self.temp_dir = Path(tempfile.mkdtemp()) - - def tearDown(self): - shutil.rmtree(self.temp_dir, ignore_errors=True) - super().tearDown() - - def test_semver_and_constraints(self): - """Test semantic version parsing, comparison, and constraint evaluation.""" - v1 = cw.SemVer.parse("1.2.3") - v2 = cw.SemVer.parse("1.2.4") - v3 = cw.SemVer.parse("2.0.0") - self.assertTrue(v1 < v2 < v3) - self.assertTrue(cw.version_satisfies(v1, ">=1.0.0,<2.0.0")) - self.assertTrue(cw.version_satisfies(v1, "==1.2.3")) - self.assertFalse(cw.version_satisfies(v1, ">1.2.3")) - - def test_schema_validator(self): - """Test deterministic JSON schema validator subset.""" - schema = { - "type": "object", - "required": ["name", "age"], - "properties": { - "name": { - "type": "string", - "minLength": 2, - "pattern": r"^[A-Z][a-z]+$", - }, - "age": {"type": "integer", "minimum": 0, "maximum": 120}, - "role": {"type": "string", "enum": ["admin", "user"]}, - "tags": {"type": "array", "items": {"type": "string"}}, - }, - "additionalProperties": False, - } - # Valid instance - self.assertEqual( - cw.validate_instance( - {"name": "Alice", "age": 30, "role": "admin", "tags": ["lead"]}, - schema, - ), - [], - ) - - # Missing required property - errs = cw.validate_instance({"name": "Alice"}, schema) - self.assertIn("$/age: required property is missing", errs) - - # Type discrimination: boolean vs integer - errs = cw.validate_instance({"name": "Alice", "age": True}, schema) - self.assertIn("$/age: expected integer, got bool", errs) - - # Pattern failure - errs = cw.validate_instance({"name": "alice123", "age": 30}, schema) - self.assertIn("$/name: string does not match '^[A-Z][a-z]+$'", errs) - - # Additional property rejection - errs = cw.validate_instance( - {"name": "Alice", "age": 30, "extra": 123}, schema - ) - self.assertIn("$/extra: additional property is not allowed", errs) - - def test_toml_manager(self): - """Test TOML section and value manipulation.""" - toml_file = self.temp_dir / "config.toml" - toml_file.write_text('[section1]\nfoo = "bar"\n', encoding="utf-8") - - # Apply setting in existing section - record = cw.TomlConfigManager.apply_set( - toml_file, "section1", "key1", "val1" - ) - self.assertIsNotNone(record) - - # Apply setting in new section - record2 = cw.TomlConfigManager.apply_set( - toml_file, "section2", "key2", True - ) - self.assertIsNotNone(record2) - - # Verify content - text = toml_file.read_text(encoding="utf-8") - self.assertIn('key1 = "val1"', text) - self.assertIn("[section2]", text) - self.assertIn("key2 = true", text) - - def test_claude_lifecycle(self): - """Test Claude Code package translation, install, and artifact flow.""" - pkg_dir = self.temp_dir / "my-pkg" - pkg_dir.mkdir() - - manifest = { - "name": "my-pkg", - "version": "1.0.0", - "description": "Test Package", - "entrypoint": { - "name": "main-entry", - "instructions": "instructions/entry.md", - "delegates": ["helper"], - }, - "agents": [{ - "name": "helper", - "description": "Helper agent", - "instructions": "instructions/helper.md", - "accepts": "schemas/input.json", - "produces": "schemas/output.json", - "delegates": [], - "tools": ["read", "search"], - }], - "copy": ["schemas", "environment"], - "environment": { - "questions": "environment/questions.json", - "schema": "schemas/env.json", - }, - "compatibility_targets": [{ - "name": "claude", - "harness": "claude-code", - "versions": ">=1.0.0", - "capabilities": {}, - }], - "schemas": [ - "schemas/input.json", - "schemas/output.json", - "schemas/env.json", - ], - } - (pkg_dir / "package.json").write_text( - json.dumps(manifest), encoding="utf-8" - ) - (pkg_dir / "instructions").mkdir() - (pkg_dir / "instructions" / "entry.md").write_text( - "Call {{agent:helper}} now.", encoding="utf-8" - ) - (pkg_dir / "instructions" / "helper.md").write_text( - "I am helper.", encoding="utf-8" - ) - (pkg_dir / "schemas").mkdir() - (pkg_dir / "schemas" / "input.json").write_text( - json.dumps({"type": "object"}), encoding="utf-8" - ) - (pkg_dir / "schemas" / "output.json").write_text( - json.dumps({"type": "object"}), encoding="utf-8" - ) - (pkg_dir / "schemas" / "env.json").write_text( - json.dumps({ - "type": "object", - "properties": {"api_key": {"type": "string"}}, - "required": ["api_key"], - }), - encoding="utf-8", - ) - (pkg_dir / "environment").mkdir() - (pkg_dir / "environment" / "questions.json").write_text( - json.dumps({ - "questions": [ - {"name": "api_key", "prompt": "Enter API Key", "type": "string"} - ] - }), - encoding="utf-8", - ) - - # 1. Verify package - cw.verify_package(pkg_dir) - - # 2. Translate package - dist_dir = self.temp_dir / "dist" - cw.translate(pkg_dir, "claude-code", dist_dir) - self.assertTrue((dist_dir / "coworker-build.json").exists()) - self.assertTrue((dist_dir / "coworker-install.json").exists()) - self.assertTrue( - (dist_dir / "project" / "skills" / "main-entry" / "SKILL.md").exists() - ) - self.assertTrue((dist_dir / "project" / "agents" / "helper.md").exists()) - - # 3. Install package - workspace_dir = self.temp_dir / "workspace" - workspace_dir.mkdir() - answers_file = self.temp_dir / "answers.json" - answers_file.write_text( - json.dumps({"api_key": "secret123"}), encoding="utf-8" - ) - - cw.install( - distribution=dist_dir, - destination=workspace_dir, - answers_path=answers_file, - non_interactive=True, - upgrade=False, - skip_harness_check=True, - ) - - env_file = workspace_dir / ".coworker" / "my-pkg" / "environment.json" - self.assertTrue(env_file.exists()) - env_data = json.loads(env_file.read_text(encoding="utf-8")) - self.assertEqual(env_data["api_key"], "secret123") - self.assertEqual(env_data["revision"], 1) - - # 4. Start-run - run_data = cw.start_run(workspace_dir, "my-pkg") - run_id = run_data["run_id"] - - # 5. Describe artifact - artifact_dir = ( - workspace_dir / ".coworker" / "my-pkg" / "runs" / run_id / "artifacts" - ) - artifact_file = artifact_dir / "result.json" - artifact_file.write_text(json.dumps({"status": "ok"}), encoding="utf-8") - - descriptor = cw.describe_artifact( - workspace=workspace_dir, - package="my-pkg", - run_id=run_id, - file="result.json", - schema="schemas/output.json", - media_type="application/json", - ) - self.assertEqual(descriptor["run_id"], run_id) - self.assertEqual(descriptor["media_type"], "application/json") - self.assertTrue(descriptor["uri"].endswith("result.json")) - - # 6. Uninstall - cw.uninstall(workspace_dir, "my-pkg") - self.assertFalse( - (workspace_dir / ".coworker" / "my-pkg" / "ownership.json").exists() - ) - - def test_codex_lifecycle_and_rollback(self): - """Test Codex package translation, install, and modified file preservation.""" - pkg_dir = self.temp_dir / "codex-pkg" - pkg_dir.mkdir() - - manifest = { - "name": "codex-pkg", - "version": "0.1.0", - "description": "Codex Package", - "entrypoint": { - "name": "orchestrator", - "instructions": "instructions/entry.md", - "delegates": ["worker"], - }, - "agents": [{ - "name": "worker", - "description": "Worker agent", - "instructions": "instructions/worker.md", - "accepts": "schemas/in.json", - "produces": "schemas/out.json", - "delegates": [], - "tools": ["write", "edit"], - }], - "copy": ["schemas", "environment"], - "environment": { - "questions": "environment/questions.json", - "schema": "schemas/env.json", - }, - "compatibility_targets": [{ - "name": "codex", - "harness": "codex", - "versions": ">=0.1.0", - "capabilities": {}, - }], - "schemas": ["schemas/in.json", "schemas/out.json", "schemas/env.json"], - } - (pkg_dir / "package.json").write_text( - json.dumps(manifest), encoding="utf-8" - ) - (pkg_dir / "instructions").mkdir() - (pkg_dir / "instructions" / "entry.md").write_text( - "Delegate to {{agent:worker}}.", encoding="utf-8" - ) - (pkg_dir / "instructions" / "worker.md").write_text( - "Working.", encoding="utf-8" - ) - (pkg_dir / "schemas").mkdir() - (pkg_dir / "schemas" / "in.json").write_text( - json.dumps({"type": "object"}), encoding="utf-8" - ) - (pkg_dir / "schemas" / "out.json").write_text( - json.dumps({"type": "object"}), encoding="utf-8" - ) - (pkg_dir / "schemas" / "env.json").write_text( - json.dumps({"type": "object"}), encoding="utf-8" - ) - (pkg_dir / "environment").mkdir() - (pkg_dir / "environment" / "questions.json").write_text( - json.dumps({"questions": []}), encoding="utf-8" - ) - - # 1. Translate - dist_dir = self.temp_dir / "dist_codex" - cw.translate(pkg_dir, "codex", dist_dir) - self.assertTrue((dist_dir / "agents" / "worker.toml").exists()) - worker_toml = (dist_dir / "agents" / "worker.toml").read_text( - encoding="utf-8" - ) - self.assertIn('sandbox_mode = "workspace-write"', worker_toml) - - # 2. Install - workspace_dir = self.temp_dir / "workspace_codex" - workspace_dir.mkdir() - cw.install( - distribution=dist_dir, - destination=workspace_dir, - answers_path=None, - non_interactive=True, - upgrade=False, - skip_harness_check=True, - ) - - codex_config = workspace_dir / ".codex" / "config.toml" - self.assertTrue(codex_config.exists()) - self.assertIn("enabled = true", codex_config.read_text(encoding="utf-8")) - - # Modify a file to test preservation on uninstall - worker_target = workspace_dir / ".codex" / "agents" / "worker.toml" - self.assertTrue(worker_target.exists()) - worker_target.write_text( - "# USER MODIFIED\n" + worker_target.read_text(encoding="utf-8"), - encoding="utf-8", - ) - - # 3. Uninstall - modified = cw.uninstall(workspace_dir, "codex-pkg") - self.assertIn(".codex/agents/worker.toml", modified) - self.assertTrue(worker_target.exists()) - - def test_cycle_and_depth_detection(self): - """Test delegation graph cycle detection and max-depth enforcement.""" - pkg_dir = self.temp_dir / "cycle-pkg" - pkg_dir.mkdir() - - # Cycle - manifest_cycle = { - "name": "cycle-pkg", - "version": "1.0.0", - "description": "Cycle", - "entrypoint": { - "name": "entry", - "instructions": "entry.md", - "delegates": ["a"], - }, - "agents": [ - { - "name": "a", - "description": "a", - "instructions": "a.md", - "accepts": "s.json", - "produces": "s.json", - "delegates": ["b"], - "tools": ["read"], - }, - { - "name": "b", - "description": "b", - "instructions": "b.md", - "accepts": "s.json", - "produces": "s.json", - "delegates": ["a"], - "tools": ["read"], - }, - ], - "environment": {"questions": "q.json", "schema": "s.json"}, - "compatibility_targets": [ - {"name": "c", "harness": "claude-code", "versions": ">=1.0.0"} - ], - } - (pkg_dir / "package.json").write_text( - json.dumps(manifest_cycle), encoding="utf-8" - ) - (pkg_dir / "entry.md").write_text("entry", encoding="utf-8") - (pkg_dir / "a.md").write_text("a", encoding="utf-8") - (pkg_dir / "b.md").write_text("b", encoding="utf-8") - (pkg_dir / "s.json").write_text("{}", encoding="utf-8") - (pkg_dir / "q.json").write_text( - json.dumps({"questions": []}), encoding="utf-8" - ) - - with self.assertRaises(cw.CoworkerError) as ctx: - cw.load_package(pkg_dir) - self.assertIn("cycle", str(ctx.exception).lower()) - - # Depth > 2: entry -> a -> b -> c - manifest_depth = { - "name": "depth-pkg", - "version": "1.0.0", - "description": "Depth", - "entrypoint": { - "name": "entry", - "instructions": "entry.md", - "delegates": ["a"], - }, - "agents": [ - { - "name": "a", - "description": "a", - "instructions": "a.md", - "accepts": "s.json", - "produces": "s.json", - "delegates": ["b"], - "tools": ["read"], - }, - { - "name": "b", - "description": "b", - "instructions": "b.md", - "accepts": "s.json", - "produces": "s.json", - "delegates": ["c"], - "tools": ["read"], - }, - { - "name": "c", - "description": "c", - "instructions": "c.md", - "accepts": "s.json", - "produces": "s.json", - "delegates": [], - "tools": ["read"], - }, - ], - "environment": {"questions": "q.json", "schema": "s.json"}, - "compatibility_targets": [ - {"name": "c", "harness": "claude-code", "versions": ">=1.0.0"} - ], - } - (pkg_dir / "package.json").write_text( - json.dumps(manifest_depth), encoding="utf-8" - ) - (pkg_dir / "c.md").write_text("c", encoding="utf-8") - - with self.assertRaises(cw.CoworkerError) as ctx: - cw.load_package(pkg_dir) - self.assertIn("two levels", str(ctx.exception).lower()) - - def test_instruction_frontmatter_rejection(self): - """Test that instruction markdown files with YAML frontmatter are rejected.""" - pkg_dir = self.temp_dir / "frontmatter_pkg" - pkg_dir.mkdir(parents=True, exist_ok=True) - manifest = { - "name": "frontmatter-pkg", - "version": "1.0.0", - "description": "Test frontmatter rejection", - "entrypoint": { - "name": "frontmatter-pkg", - "instructions": "entry.md", - "delegates": [], - }, - "agents": [], - "environment": {"questions": "q.json", "schema": "s.json"}, - "compatibility_targets": [ - {"name": "c", "harness": "claude-code", "versions": ">=1.0.0"} - ], - } - (pkg_dir / "package.json").write_text(json.dumps(manifest), encoding="utf-8") - (pkg_dir / "q.json").write_text("{}", encoding="utf-8") - (pkg_dir / "s.json").write_text("{}", encoding="utf-8") - (pkg_dir / "entry.md").write_text( - "---\nname: entry\ndescription: invalid\n---\nPrompt body", - encoding="utf-8", - ) - - with self.assertRaises(cw.CoworkerError) as ctx: - cw.load_package(pkg_dir) - self.assertIn("must not contain yaml frontmatter", str(ctx.exception).lower()) - - def test_translate_runs_validation(self): - """Test that cw.translate runs package verification first and fails on invalid packages.""" - pkg_dir = self.temp_dir / "invalid_translate_pkg" - pkg_dir.mkdir(parents=True, exist_ok=True) - manifest = { - "name": "invalid-translate-pkg", - "version": "1.0.0", - "description": "Test translate validation", - "entrypoint": { - "name": "invalid-translate-pkg", - "instructions": "entry.md", - "delegates": [], - }, - "agents": [], - "environment": {"questions": "q.json", "schema": "s.json"}, - "compatibility_targets": [ - {"name": "c", "harness": "claude-code", "versions": ">=1.0.0"} - ], - "schemas": ["invalid_schema.json"], - } - (pkg_dir / "package.json").write_text(json.dumps(manifest), encoding="utf-8") - (pkg_dir / "q.json").write_text("{}", encoding="utf-8") - (pkg_dir / "s.json").write_text("{}", encoding="utf-8") - (pkg_dir / "entry.md").write_text("Valid prompt body", encoding="utf-8") - (pkg_dir / "invalid_schema.json").write_text( - '{"type": "invalid_type"}', encoding="utf-8" - ) - - output_dir = self.temp_dir / "out" - with self.assertRaises(cw.CoworkerError) as ctx: - cw.translate(pkg_dir, "claude-code", output_dir) - self.assertIn("invalid schema", str(ctx.exception).lower()) - - def test_cli_main(self): - """Test CLI validate subcommand with valid and invalid JSON inputs.""" - schema_path = self.temp_dir / "schema.json" - schema_path.write_text(json.dumps({"type": "integer"}), encoding="utf-8") - valid_instance = self.temp_dir / "valid.json" - valid_instance.write_text("42", encoding="utf-8") - invalid_instance = self.temp_dir / "invalid.json" - invalid_instance.write_text('"forty-two"', encoding="utf-8") - - ret_ok = cw.main( - ["validate", "--schema", str(schema_path), str(valid_instance)] - ) - self.assertEqual(ret_ok, 0) - - ret_invalid = cw.main( - ["validate", "--schema", str(schema_path), str(invalid_instance)] - ) - self.assertEqual(ret_invalid, 2) - - -if __name__ == "__main__": - unittest.main() - diff --git a/nexus/framework/tests/test_installer.py b/nexus/framework/tests/test_installer.py new file mode 100644 index 0000000..ea3a01a --- /dev/null +++ b/nexus/framework/tests/test_installer.py @@ -0,0 +1,192 @@ +"""Unit tests for installer module (install, upgrade, rollback, uninstall).""" + +import json +import pathlib +import shutil +import tempfile +import unittest + +try: + from accelerator_agents.tpu_nexus.framework import compiler, installer, utils +except ImportError: + try: + from .. import compiler, installer, utils + except ImportError: + import compiler + import installer + import utils + +Path = pathlib.Path + + +class TestInstaller(unittest.TestCase): + """Unit tests for installer functions, pre-validations, collisions, and rollback.""" + + def setUp(self): + super().setUp() + self.temp_dir = Path(tempfile.mkdtemp()) + self.pkg_dir = self._create_test_package() + self.dist_dir = self.temp_dir / "dist" + compiler.translate(self.pkg_dir, "claude-code", self.dist_dir) + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + super().tearDown() + + def _create_test_package(self) -> Path: + pkg_dir = self.temp_dir / "pkg" + pkg_dir.mkdir(parents=True, exist_ok=True) + manifest = { + "name": "pkg", + "version": "1.0.0", + "description": "Installer test package", + "entrypoint": { + "name": "entry", + "instructions": "instructions/entry.md", + "delegates": ["helper"], + }, + "agents": [{ + "name": "helper", + "description": "Helper", + "instructions": "instructions/helper.md", + "accepts": "schemas/in.json", + "produces": "schemas/out.json", + "delegates": [], + "tools": ["read"], + }], + "copy": ["schemas", "environment"], + "environment": { + "questions": "environment/questions.json", + "schema": "schemas/env.json", + }, + "compatibility_targets": [{ + "name": "claude", + "harness": "claude-code", + "versions": ">=2.0.0", + "capabilities": {}, + }], + "schemas": ["schemas/in.json", "schemas/out.json", "schemas/env.json"], + } + (pkg_dir / "package.json").write_text( + json.dumps(manifest), encoding="utf-8" + ) + (pkg_dir / "instructions").mkdir(parents=True, exist_ok=True) + (pkg_dir / "instructions" / "entry.md").write_text( + "Call {{agent:helper}}.", encoding="utf-8" + ) + (pkg_dir / "instructions" / "helper.md").write_text( + "Help.", encoding="utf-8" + ) + (pkg_dir / "schemas").mkdir(parents=True, exist_ok=True) + (pkg_dir / "schemas" / "in.json").write_text( + json.dumps({"type": "object"}), encoding="utf-8" + ) + (pkg_dir / "schemas" / "out.json").write_text( + json.dumps({"type": "object"}), encoding="utf-8" + ) + (pkg_dir / "schemas" / "env.json").write_text( + json.dumps({ + "type": "object", + "properties": { + "tpu_target": {"type": "string"}, + "timeout": {"type": "integer"}, + }, + "required": ["tpu_target", "timeout"], + }), + encoding="utf-8", + ) + (pkg_dir / "environment").mkdir(parents=True, exist_ok=True) + (pkg_dir / "environment" / "questions.json").write_text( + json.dumps({ + "questions": [ + { + "name": "tpu_target", + "prompt": "TPU target", + "type": "string", + "default": "local-tpu", + }, + { + "name": "timeout", + "prompt": "Timeout", + "type": "integer", + "default": 30, + }, + ] + }), + encoding="utf-8", + ) + return pkg_dir + + def test_install_and_environment_resolution(self): + workspace = self.temp_dir / "workspace" + workspace.mkdir() + + answers = self.temp_dir / "answers.json" + answers.write_text( + json.dumps({"tpu_target": "node-1", "timeout": 60}), encoding="utf-8" + ) + + installer.install( + distribution=self.dist_dir, + destination=workspace, + answers_path=answers, + non_interactive=True, + upgrade=False, + skip_harness_check=True, + ) + + env_file = workspace / ".coworker" / "pkg" / "environment.json" + self.assertTrue(env_file.is_file()) + env = json.loads(env_file.read_text(encoding="utf-8")) + self.assertEqual(env["tpu_target"], "node-1") + self.assertEqual(env["timeout"], 60) + self.assertEqual(env["revision"], 1) + + ownership_file = workspace / ".coworker" / "pkg" / "ownership.json" + self.assertTrue(ownership_file.is_file()) + + def test_collision_detection(self): + workspace = self.temp_dir / "workspace_collision" + workspace.mkdir() + (workspace / ".claude" / "skills" / "entry").mkdir(parents=True) + (workspace / ".claude" / "skills" / "entry" / "SKILL.md").write_text( + "EXISTING UNMANAGED FILE", encoding="utf-8" + ) + + with self.assertRaises(utils.CoworkerError) as ctx: + installer.install( + distribution=self.dist_dir, + destination=workspace, + answers_path=None, + non_interactive=True, + upgrade=False, + skip_harness_check=True, + ) + self.assertIn("refusing to overwrite", str(ctx.exception).lower()) + + def test_uninstall_and_rollback(self): + workspace = self.temp_dir / "workspace_uninstall" + workspace.mkdir() + + installer.install( + distribution=self.dist_dir, + destination=workspace, + answers_path=None, + non_interactive=True, + upgrade=False, + skip_harness_check=True, + ) + + # Modify an agent file + agent_file = workspace / ".claude" / "agents" / "helper.md" + self.assertTrue(agent_file.is_file()) + agent_file.write_text("# USER MODIFIED", encoding="utf-8") + + modified = installer.uninstall(workspace, "pkg") + self.assertIn(".claude/agents/helper.md", modified) + self.assertTrue(agent_file.is_file()) # Preserved + self.assertFalse((workspace / ".coworker" / "pkg" / "ownership.json").exists()) + + +if __name__ == "__main__": + unittest.main() diff --git a/nexus/framework/tests/test_runtime.py b/nexus/framework/tests/test_runtime.py new file mode 100644 index 0000000..f352ec2 --- /dev/null +++ b/nexus/framework/tests/test_runtime.py @@ -0,0 +1,104 @@ +"""Unit tests for standalone runtime operations.""" + +import json +import pathlib +import shutil +import tempfile +import unittest + +try: + from accelerator_agents.tpu_nexus.framework import runtime, utils +except ImportError: + try: + from .. import runtime, utils + except ImportError: + import runtime + import utils + +Path = pathlib.Path + + +class TestRuntime(unittest.TestCase): + """Unit tests for runtime namespace allocation, descriptors, and validations.""" + + def setUp(self): + super().setUp() + self.temp_dir = Path(tempfile.mkdtemp()) + self.workspace_dir = self.temp_dir / "workspace" + self.workspace_dir.mkdir() + + # Set up installed package structure + self.pkg_root = self.workspace_dir / ".coworker" / "test-pkg" + self.pkg_root.mkdir(parents=True) + (self.pkg_root / "environment.json").write_text( + json.dumps({"revision": 1, "target": "local"}), encoding="utf-8" + ) + (self.pkg_root / "schemas").mkdir() + (self.pkg_root / "schemas" / "out.json").write_text( + json.dumps({ + "type": "object", + "properties": {"status": {"type": "string"}}, + "required": ["status"], + }), + encoding="utf-8", + ) + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + super().tearDown() + + def test_start_run_and_namespace(self): + doc = runtime.start_run(self.workspace_dir, "test-pkg") + self.assertIn("run_id", doc) + self.assertEqual(doc["package"], "test-pkg") + self.assertEqual(doc["environment_revision"], 1) + + run_dir = self.pkg_root / "runs" / doc["run_id"] + self.assertTrue(run_dir.is_dir()) + self.assertTrue((run_dir / "artifacts").is_dir()) + self.assertTrue((run_dir / "run.json").is_file()) + + def test_describe_artifact(self): + run_doc = runtime.start_run(self.workspace_dir, "test-pkg") + run_id = run_doc["run_id"] + + artifact_file = ( + self.pkg_root / "runs" / run_id / "artifacts" / "sub" / "output.json" + ) + artifact_file.parent.mkdir(parents=True, exist_ok=True) + artifact_file.write_text(json.dumps({"status": "ok"}), encoding="utf-8") + + desc = runtime.describe_artifact( + workspace=self.workspace_dir, + package="test-pkg", + run_id=run_id, + file="sub/output.json", + schema="schemas/out.json", + media_type="application/json", + ) + + self.assertEqual(desc["run_id"], run_id) + self.assertEqual(desc["media_type"], "application/json") + self.assertEqual(desc["schema"], "schemas/out.json") + self.assertEqual( + desc["uri"], + f"workspace://test-pkg/runs/{run_id}/artifacts/sub/output.json", + ) + self.assertEqual(desc["sha256"], utils.compute_digest(artifact_file)) + + def test_runtime_cli(self): + run_doc = runtime.start_run(self.workspace_dir, "test-pkg") + run_id = run_doc["run_id"] + + valid_instance = self.temp_dir / "valid.json" + valid_instance.write_text(json.dumps({"status": "ok"}), encoding="utf-8") + schema_path = self.pkg_root / "schemas" / "out.json" + + ret = runtime.main( + ["validate", "--schema", str(schema_path), str(valid_instance)] + ) + self.assertEqual(ret, 0) + + +if __name__ == "__main__": + unittest.main() diff --git a/nexus/framework/tests/test_utils.py b/nexus/framework/tests/test_utils.py new file mode 100644 index 0000000..9c79062 --- /dev/null +++ b/nexus/framework/tests/test_utils.py @@ -0,0 +1,115 @@ +"""Unit tests for shared utils module.""" + +import json +import pathlib +import shutil +import tempfile +import unittest + +try: + from accelerator_agents.tpu_nexus.framework import utils +except ImportError: + try: + from .. import utils + except ImportError: + import utils + +Path = pathlib.Path + + +class TestUtils(unittest.TestCase): + """Unit tests for utils functions and config managers.""" + + def setUp(self): + super().setUp() + self.temp_dir = Path(tempfile.mkdtemp()) + + def tearDown(self): + shutil.rmtree(self.temp_dir, ignore_errors=True) + super().tearDown() + + def test_semver_parsing_and_ranges(self): + v1 = utils.SemVer.parse("1.2.3") + v2 = utils.SemVer.parse("1.2.4") + v3 = utils.SemVer.parse("2.0.0") + self.assertTrue(v1 < v2 < v3) + self.assertTrue(utils.version_satisfies(v1, ">=1.0.0,<2.0.0")) + self.assertTrue(utils.version_satisfies(v1, "==1.2.3")) + self.assertFalse(utils.version_satisfies(v1, ">1.2.3")) + self.assertTrue(utils.version_satisfies(v2, ">=1.2.0")) + + def test_schema_validator(self): + schema = { + "type": "object", + "required": ["name", "age"], + "properties": { + "name": { + "type": "string", + "minLength": 2, + "pattern": r"^[A-Z][a-z]+$", + }, + "age": {"type": "integer", "minimum": 0, "maximum": 120}, + "role": {"type": "string", "enum": ["admin", "user"]}, + "tags": {"type": "array", "items": {"type": "string"}}, + }, + "additionalProperties": False, + } + # Valid instance + self.assertEqual( + utils.validate_instance( + {"name": "Alice", "age": 30, "role": "admin", "tags": ["lead"]}, + schema, + ), + [], + ) + + # Missing required property + errs = utils.validate_instance({"name": "Alice"}, schema) + self.assertIn("$/age: required property is missing", errs) + + # Type discrimination: boolean vs integer + errs = utils.validate_instance({"name": "Alice", "age": True}, schema) + self.assertIn("$/age: expected integer, got bool", errs) + + # Pattern failure + errs = utils.validate_instance({"name": "alice123", "age": 30}, schema) + self.assertIn("$/name: string does not match '^[A-Z][a-z]+$'", errs) + + # Additional property rejection + errs = utils.validate_instance( + {"name": "Alice", "age": 30, "extra": 123}, schema + ) + self.assertIn("$/extra: additional property is not allowed", errs) + + def test_toml_manager(self): + toml_file = self.temp_dir / "config.toml" + toml_file.write_text('[section1]\nfoo = "bar"\n', encoding="utf-8") + + # Apply setting in existing section + record = utils.TomlConfigManager.apply_set( + toml_file, "section1", "key1", "val1" + ) + self.assertIsNotNone(record) + + # Apply setting in new section + record2 = utils.TomlConfigManager.apply_set( + toml_file, "section2", "key2", True + ) + self.assertIsNotNone(record2) + + # Verify content + text = toml_file.read_text(encoding="utf-8") + self.assertIn('key1 = "val1"', text) + self.assertIn("[section2]", text) + self.assertIn("key2 = true", text) + + def test_json_config_manager(self): + doc = {"a": {"b": 1}} + records = [] + utils.JsonConfigManager.merge(doc, {"a": {"c": 2}, "d": 3}, [], records) + self.assertEqual(doc, {"a": {"b": 1, "c": 2}, "d": 3}) + self.assertEqual(len(records), 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/nexus/package/maxkernel_package_converter.py b/nexus/package/maxkernel_package_converter.py index 80a7445..94ce6d8 100644 --- a/nexus/package/maxkernel_package_converter.py +++ b/nexus/package/maxkernel_package_converter.py @@ -725,7 +725,7 @@ def main() -> None: ) print("\nConversion complete! You can verify the package with:") - print(f" python3 framework/coworker.py verify {target_dir}") + print(f" python3 framework/compiler.py verify {target_dir}") if __name__ == "__main__":