Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ jobs:
- name: Verify routing, observation, process, guarded, and conformance boundaries
run: |
PYTHONPATH=src python -W error::ResourceWarning -m unittest \
tests.test_router tests.test_native_v2_runtime \
tests.test_router tests.test_agent_discovery tests.test_native_v2_runtime \
tests.test_executable_observation tests.test_foreground_process \
tests.test_process_context tests.test_triage tests.test_json_input \
tests.test_cost_recommendation \
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/release.yml
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,7 @@ jobs:
- name: Verify routing, observation, process, guarded, and conformance boundaries
run: |
PYTHONPATH=src python -W error::ResourceWarning -m unittest \
tests.test_router tests.test_native_v2_runtime \
tests.test_router tests.test_agent_discovery tests.test_native_v2_runtime \
tests.test_executable_observation tests.test_foreground_process \
tests.test_process_context tests.test_triage tests.test_json_input \
tests.test_cost_recommendation \
Expand Down
70 changes: 68 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -68,10 +68,11 @@ ownership labels remain opaque caller declarations. See the
`wclass --help` lists the whole surface:

```text
wclass [-h] [--version] {classify,example-policy,review-preset,review-cost-profile,recommend,route,run,render,delegate,v2} ...
wclass [-h] [--version] {discover,profile,classify,example-policy,review-preset,review-cost-profile,recommend,route,run,render,delegate,v2} ...
```

`classify`, `recommend`, `route`, and `run` read the task from standard input. `render`
`classify`, `recommend`, `route`, and `run` read the task from standard input. `discover`
and `profile` are task-free local selection commands. `render`
prints the command of a policy route named by a workflow descriptor and never
reads a task. `example-policy` emits packaged policy JSON; `review-preset`
prints every command and fingerprint in one packaged policy.
Expand Down Expand Up @@ -106,6 +107,71 @@ them:
Code `1` is not weightclass's; it means the interpreter died on an unhandled
exception, which is a bug worth reporting.

## Discover installed agents and generate a policy

`discover` checks only absolute directories in the current `PATH` for the four
package-supported executable names. It does not start a vendor process, read
vendor configuration or authentication files, make a network request, or read
task standard input:

```sh
wclass discover
wclass discover --agent grok
```

The JSON result distinguishes an executable detected on the local path from a
usable subscription or model. `executable_detected` means only that a regular
executable file was found. Subscription, pricing, and quota remain `unknown`.
The package-owned effort catalog describes the command shapes weightclass can
build; it is not a probe of the installed CLI version. The model catalog
contains only `default`, meaning that no model override is emitted, and reports
`availability_verified: false`. Cloud model entitlement is not a locally
installed property that weightclass can safely infer.

`profile` turns an agent, model, effort, and tier selection into a complete
schema-1 policy, so the user does not have to assemble vendor argv manually:

```sh
wclass profile \
--agent codex \
--tier low \
--model default \
--effort low > worker-policy.json
```

Codex, Claude, and Grok accept an opaque `--model` selection through their
closed package builders. `agy` currently accepts only `--model default`
because weightclass has no reviewed model-override shape for it. Every
non-default model label remains caller-supplied opaque configuration;
weightclass does not verify that the account can use it. The generated policy
contains the detected absolute executable path and exactly one tier route.
The command writes nothing unless the caller explicitly redirects its output
to a chosen file.

For an intentional cross-vendor worker, add `--allow-cross-vendor`. This emits
the existing schema-1 `allow_mixed_vendors: true` opt-in; it is deliberately
not a directional grant, so use the generated single-worker policy only at the
reviewed integration boundary:

```sh
wclass profile \
--agent grok \
--tier low \
--model default \
--effort low \
--allow-cross-vendor > worker-policy.json

printf '%s' 'Fix a spelling typo.' | \
wclass route --policy worker-policy.json --source-vendor codex --tier low
```

Review the emitted route and pass its fingerprint to the ordinary `run`
command. Discovery and profile generation never execute the selected agent;
`run` still starts exactly one foreground child with no retry or fallback.
Generated `agy` and Grok policies retain `task_delivery: argv` and its local
process-inspection exposure. Schema 1 binds the lexical executable path in the
route fingerprint but does not provide schema-2 executable reobservation.

Code `7` carries the real status in its diagnostic, as
`{"error": "executor_failed", "executor_exit_code": N}` or, for a command killed
by a signal, `{"error": "executor_failed", "executor_signal": N}`. A selected
Expand Down
21 changes: 21 additions & 0 deletions docs/integrations.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,27 @@ the project directory after installing `weightclass`.
3. Keep a Codex-originated task on Codex and a Claude-originated task on Claude
unless a reviewed policy explicitly enables cross-vendor routing.

## Local discovery and profile selection

Use `wclass discover` to list package-supported agent executable names found in
absolute `PATH` entries. It performs filesystem checks only: no agent is
started, no task is read, and no vendor configuration, authentication file, or
network service is accessed. The result does not prove subscription access,
model entitlement, price, quota, or installed-CLI compatibility. Model and
effort catalogs are package declarations with availability explicitly
unverified.

Use `wclass profile --agent <agent> --tier <tier> --model <label> --effort
<effort>` to generate one schema-1 policy on standard output. `default` omits a
model override; another accepted model label is opaque user configuration, not
an availability claim. `agy` supports only `default` because it has no reviewed
model-override builder. Add `--allow-cross-vendor` only for an intentional
boundary change, then review the generated policy through the ordinary
fingerprint-bound `route`/`run` workflow. The generated policy uses the
detected absolute lexical executable path, starts nothing by itself, and is
never persisted unless the caller explicitly writes its stdout to a selected
file.

Do not put credentials, tokens, or personal information on a command line, in
a policy, or in vendor-global configuration. Task text is a deliberate
exception to that rule, not an oversight: the built-in `agy` and `grok` routes,
Expand Down
202 changes: 202 additions & 0 deletions src/weightclass/agent_discovery.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,202 @@
"""Task-free discovery of package-supported native agent executables."""

from __future__ import annotations

import os
import unicodedata
from dataclasses import dataclass
from typing import Literal

from .router import CLAUDE_COMMAND_PREFIX, agy_command, codex_command, grok_command

DiscoveryTaskDelivery = Literal["stdin", "argv"]

MAX_PATH_BYTES = 32_768
MAX_PATH_ENTRIES = 256


class AgentDiscoveryError(ValueError):
"""A value-free local discovery failure."""


class AgentUnavailableError(LookupError):
"""Raised when a selected package-supported executable is not detected."""


@dataclass(frozen=True, slots=True)
class AgentAdapter:
agent: str
executable_name: str
task_delivery: DiscoveryTaskDelivery
accepts_opaque_model_override: bool


AGENT_ADAPTERS = (
AgentAdapter("agy", "agy", "argv", False),
AgentAdapter("claude", "claude", "stdin", True),
AgentAdapter("codex", "codex", "stdin", True),
AgentAdapter("grok", "grok", "argv", True),
)
AGENT_IDS = tuple(adapter.agent for adapter in AGENT_ADAPTERS)
TIERS = ("low", "standard", "high")
EFFORTS = ("low", "medium", "high")
MAX_MODEL_LABEL_BYTES = 240


def _path_entries(path_value: str) -> tuple[str, ...]:
try:
encoded = path_value.encode("utf-8", errors="strict")
except UnicodeEncodeError:
raise AgentDiscoveryError() from None
entries = path_value.split(os.pathsep)
if len(encoded) > MAX_PATH_BYTES or len(entries) > MAX_PATH_ENTRIES:
raise AgentDiscoveryError()
return tuple(entry for entry in entries if entry and os.path.isabs(entry))


def _reviewable_path(value: str) -> bool:
return value == value.strip(" ") and not any(
unicodedata.category(character).startswith("C")
or (character.isspace() and character != " ")
for character in value
)


def _find_executable(name: str, entries: tuple[str, ...]) -> str | None:
for directory in entries:
candidate = os.path.join(directory, name)
if not _reviewable_path(candidate):
continue
try:
if os.path.isfile(candidate) and os.access(candidate, os.X_OK):
return candidate
except OSError:
continue
return None


def render_agent_discovery(
path_value: str | None = None,
*,
agent: str | None = None,
) -> dict[str, object]:
"""Describe locally detected built-in agents without starting a process."""
if agent is not None and agent not in AGENT_IDS:
raise AgentDiscoveryError()
entries = _path_entries(os.environ.get("PATH", "") if path_value is None else path_value)
agents: list[dict[str, object]] = []
for adapter in AGENT_ADAPTERS:
if agent is not None and adapter.agent != agent:
continue
executable = _find_executable(adapter.executable_name, entries)
agents.append(
{
"agent": adapter.agent,
"executable": executable,
"executable_detected": executable is not None,
"task_delivery": adapter.task_delivery,
"model_catalog": {
"source": "package_default_only",
"values": ["default"],
"accepts_opaque_override": adapter.accepts_opaque_model_override,
"availability_verified": False,
},
"effort_catalog": {
"source": "package_catalog",
"values": ["low", "medium", "high"],
"availability_verified": False,
},
"subscription": "unknown",
"pricing": "unknown",
"quota": "unknown",
}
)
return {
"schema_version": 1,
"discovery_mode": "local_path_only",
"network_used": False,
"vendor_processes_started": False,
"agents": agents,
}


def _model_label(value: object) -> str:
if not isinstance(value, str):
raise AgentDiscoveryError()
try:
encoded = value.encode("utf-8", errors="strict")
except UnicodeEncodeError:
raise AgentDiscoveryError() from None
if (
not 1 <= len(encoded) <= MAX_MODEL_LABEL_BYTES
or value.startswith("-")
or any(character.isspace() or not character.isprintable() for character in value)
):
raise AgentDiscoveryError()
return value


def _selected_command(
adapter: AgentAdapter,
executable: str,
model: str,
effort: str,
) -> list[str]:
if adapter.agent == "codex":
command = list(codex_command(effort))
if model != "default":
command[command.index("-c") : command.index("-c")] = ["--model", model]
elif adapter.agent == "claude":
command = list(CLAUDE_COMMAND_PREFIX + (effort,))
if model != "default":
insertion = command.index("--effort")
command[insertion:insertion] = ["--model", model]
elif adapter.agent == "agy":
if model != "default":
raise AgentDiscoveryError()
command = list(agy_command(effort))
elif adapter.agent == "grok":
command = list(grok_command(effort))
if model != "default":
insertion = command.index("--reasoning-effort")
command[insertion:insertion] = ["--model", model]
else:
raise AgentDiscoveryError()
command[0] = executable
return command


def generate_selected_policy(
*,
agent: str,
tier: str,
model: str,
effort: str,
allow_cross_vendor: bool,
path_value: str | None = None,
) -> dict[str, object]:
"""Compile one selected built-in agent profile into a schema-1 policy."""
if tier not in TIERS or effort not in EFFORTS or not isinstance(allow_cross_vendor, bool):
raise AgentDiscoveryError()
selected_model = _model_label(model)
adapter = next((item for item in AGENT_ADAPTERS if item.agent == agent), None)
if adapter is None:
raise AgentDiscoveryError()
entries = _path_entries(os.environ.get("PATH", "") if path_value is None else path_value)
executable = _find_executable(adapter.executable_name, entries)
if executable is None:
raise AgentUnavailableError()
command = _selected_command(adapter, executable, selected_model, effort)
return {
"schema_version": 1,
"allow_mixed_vendors": allow_cross_vendor,
"posture": "balanced",
"routes": [
{
"id": f"selected-{adapter.agent}-{tier}",
"vendor": adapter.agent,
"tier": tier,
"command": command,
}
],
}
Loading