diff --git a/harness2/cli/doctor.py b/harness2/cli/doctor.py
index dbd20bf2..5da655ca 100644
--- a/harness2/cli/doctor.py
+++ b/harness2/cli/doctor.py
@@ -16,7 +16,9 @@
from __future__ import annotations
+import os
import subprocess
+from pathlib import Path
from .. import providers, registry
from ..core.proposer import Proposer
@@ -79,7 +81,7 @@ def resolve(model=model, channel=channel, role=role):
if channel == "claude":
providers.claude_env(provider)
elif channel == "codex":
- providers.codex_config(provider, bare, role["effort"])
+ providers.codex_args(provider, bare, role["effort"])
elif channel == "opencode":
providers.opencode_config(provider, bare)
if channel != "command":
@@ -128,6 +130,16 @@ def binary_check(binary=binary, channel=channel):
"detail": "; ".join(problems) or detail,
}
)
+ if any(config.provider(name).type == "chatgpt-login" for name in referenced):
+ agents = Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex") / "AGENTS.md"
+ if agents.is_file() and (text := agents.read_text(encoding="utf-8")):
+ rows.append(
+ {
+ "check": "Codex global AGENTS.md",
+ "ok": True,
+ "detail": f"your global AGENTS.md ({len(text)} characters) is part of every Codex run",
+ }
+ )
if args.live:
for (channel, model), spec in sorted(models.items()):
diff --git a/harness2/core/proposer.py b/harness2/core/proposer.py
index a0aae876..5c2d23a4 100644
--- a/harness2/core/proposer.py
+++ b/harness2/core/proposer.py
@@ -35,7 +35,7 @@
from pathlib import Path
from .. import providers
-from ..substrate.codex import parse_items_text
+from ..substrate.codex import CodexSubstrate, parse_items_text
from . import events
from .spec import ChannelSpec
@@ -165,8 +165,11 @@ def call(
tmp, mission, system, wall
)
spent_usd += cost
+ if err and providers.USAGE_LIMIT.search(err):
+ events.request_stop("usage_limit")
+ raise events.Stopped("usage_limit")
if not text.strip():
- last = err or "empty reply"
+ last = err[-300:] or "empty reply"
continue
r = Reply(
label=label,
@@ -193,6 +196,9 @@ def call(
last = f"{type(exc).__name__}: {exc}"
finally:
shutil.rmtree(tmp, ignore_errors=True)
+ if providers.USAGE_LIMIT.search(last):
+ events.request_stop("usage_limit")
+ raise events.Stopped("usage_limit")
r = Reply(
label=label,
@@ -214,8 +220,12 @@ def call(
)
def _claude(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str, float, str]:
- home = cwd / ".claude-home"
- home.mkdir()
+ login = self.spec.provider is not None and self.spec.provider.type == "claude-login"
+ env = self.env()
+ if not login:
+ home = cwd / ".claude-home"
+ home.mkdir()
+ env["CLAUDE_CONFIG_DIR"] = str(home)
cmd = [
*self._bin(),
"--dangerously-skip-permissions",
@@ -228,6 +238,7 @@ def _claude(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str,
self._model(),
"--setting-sources",
"project,local",
+ *(["--strict-mcp-config", "--no-session-persistence"] if login else []),
]
stdin = mission
if system:
@@ -243,7 +254,7 @@ def _claude(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str,
cmd,
input=stdin,
cwd=cwd,
- env=self.env() | {"CLAUDE_CONFIG_DIR": str(home)},
+ env=env,
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
@@ -252,24 +263,28 @@ def _claude(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str,
payload = _parse_json(p.stdout)
text = str(payload.get("result") or payload.get("text") or "")
err = ""
- if payload.get("is_error") or payload.get("subtype") not in (None, "success"):
- err = f"CLI error: {payload.get('subtype')} {str(text)[:200]}"
+ if (
+ p.returncode
+ or payload.get("is_error")
+ or payload.get("subtype") not in (None, "success")
+ ):
+ err = f"rc={p.returncode}: {p.stdout} {p.stderr}"
text = ""
- elif p.returncode != 0 and not text.strip():
- err = f"rc={p.returncode}: {(p.stderr or '')[-300:]}"
return text, float(payload.get("total_cost_usd") or 0.0), err
def _codex(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str, float, str]:
- home = cwd / ".codex-home"
- home.mkdir()
model = self._model()
- config = (
- providers.codex_config(self.spec.provider, model, self.spec.effort)
- if self.spec.provider is not None
- else f"model = {json.dumps(model)}\n"
- f"model_reasoning_effort = {json.dumps(self.spec.effort)}\n"
- )
- (home / "config.toml").write_text(config, encoding="utf-8")
+ home = None
+ if self.spec.provider is None or self.spec.provider.type != "chatgpt-login":
+ home = cwd / ".codex-home"
+ home.mkdir()
+ config = (
+ providers.codex_config(self.spec.provider, model, self.spec.effort)
+ if self.spec.provider is not None
+ else f"model = {json.dumps(model)}\n"
+ f"model_reasoning_effort = {json.dumps(self.spec.effort)}\n"
+ )
+ (home / "config.toml").write_text(config, encoding="utf-8")
p = subprocess.run(
[
*self._bin(),
@@ -278,13 +293,14 @@ def _codex(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str,
"--skip-git-repo-check",
"--dangerously-bypass-approvals-and-sandbox",
"--ephemeral",
+ *providers.codex_args(self.spec.provider, model, self.spec.effort),
"-C",
str(cwd),
"-",
],
input=f"# SYSTEM INSTRUCTIONS\n\n{system}\n\n# END\n\n{mission}" if system else mission,
cwd=cwd,
- env=self.env() | {"CODEX_HOME": str(home)},
+ env=CodexSubstrate().env(self.env(), home, provider=self.spec.provider),
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
@@ -299,7 +315,7 @@ def _codex(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str,
if item.get("type") == "agent_message":
text = str(item.get("text") or "")
if errors or p.returncode:
- return "", 0.0, "; ".join(errors)[-300:] or f"rc={p.returncode}: {p.stderr[-300:]}"
+ return "", 0.0, f"rc={p.returncode}: {p.stdout} {p.stderr}"
return text, 0.0, ""
def _opencode(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[str, float, str]:
@@ -348,10 +364,12 @@ def _opencode(self, cwd: Path, mission: str, system: str, wall: int) -> tuple[st
timeout=wall,
)
cost = _opencode_cost(p.stdout)
+ text, err = _opencode_stdout(p.stdout)
+ if (p.returncode or err) and providers.USAGE_LIMIT.search(p.stdout + p.stderr):
+ return "", cost, p.stdout + p.stderr
ans = out / "answer.md"
if ans.is_file() and ans.read_text(errors="replace").strip():
return ans.read_text(errors="replace"), cost, ""
- text, err = _opencode_stdout(p.stdout)
if text.strip():
return text, cost, ""
tail = _strip_oc_banner(p.stderr or "").strip()[-300:]
diff --git a/harness2/providers.py b/harness2/providers.py
index 9901ef27..96053697 100644
--- a/harness2/providers.py
+++ b/harness2/providers.py
@@ -18,12 +18,17 @@
import json
import os
+import re
+import subprocess
from dataclasses import dataclass
from urllib.parse import urlsplit
from .settings import ConfigError
-PROVIDER_TYPES = ("anthropic", "openai", "vertex", "bedrock")
+PROVIDER_TYPES = ("anthropic", "openai", "vertex", "bedrock", "claude-login", "chatgpt-login")
+USAGE_LIMIT = re.compile(
+ r"(?i)usage limit|rate[ _-]?limit|too many requests|\b429\b|quota|credit balance is too low"
+)
STRIP_ENV = (
"ANTHROPIC_API_KEY",
"ANTHROPIC_AUTH_TOKEN",
@@ -60,6 +65,10 @@ def split_model(ref: str) -> tuple[str, str]:
def claude_env(p: Provider) -> dict[str, str]:
+ if p.type == "claude-login":
+ return {}
+ if p.type == "chatgpt-login":
+ raise ConfigError("Claude Code cannot use a ChatGPT login.")
if p.type == "anthropic":
env = {}
if p.api_key_env:
@@ -104,7 +113,24 @@ def quote(value):
)
+def codex_args(p: Provider | None, model: str, effort: str) -> list[str]:
+ if p is not None and p.type == "chatgpt-login":
+ return [
+ "--ignore-user-config",
+ "--ignore-rules",
+ "-m",
+ model,
+ "-c",
+ f"model_reasoning_effort={json.dumps(effort)}",
+ ]
+ if p is not None:
+ codex_config(p, model, effort)
+ return []
+
+
def opencode_config(p: Provider, model: str) -> tuple[dict, dict[str, str]]:
+ if p.type in ("claude-login", "chatgpt-login"):
+ raise ConfigError("OpenCode needs an API provider; subscription logins are not supported.")
if p.type == "vertex":
region = p.region or "global"
return {}, {
@@ -135,6 +161,10 @@ def opencode_config(p: Provider, model: str) -> tuple[dict, dict[str, str]]:
def litellm_params(p: Provider, model: str) -> dict:
+ if p.type in ("claude-login", "chatgpt-login"):
+ raise ConfigError(
+ "The bridge needs an API provider; subscription logins are not supported."
+ )
if p.type in ("anthropic", "openai"):
params = {"model": f"{p.type}/{model}", "api_key": f"os.environ/{p.api_key_env}"}
if p.base_url or p.type == "openai":
@@ -152,6 +182,32 @@ def litellm_params(p: Provider, model: str) -> dict:
def check(p: Provider) -> list[str]:
+ if p.type in ("claude-login", "chatgpt-login"):
+ argv = (
+ ["claude", "auth", "status"]
+ if p.type == "claude-login"
+ else ["codex", "login", "status"]
+ )
+ login = "claude auth login" if p.type == "claude-login" else "codex login"
+ try:
+ result = subprocess.run(
+ argv,
+ env={
+ k: v
+ for k, v in os.environ.items()
+ if k not in STRIP_ENV
+ and (p.type != "chatgpt-login" or k not in ("OPENAI_API_KEY", "CODEX_API_KEY"))
+ },
+ stdin=subprocess.DEVNULL,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ timeout=10,
+ )
+ if result.returncode == 0:
+ return []
+ except (OSError, subprocess.SubprocessError):
+ pass
+ return [f"{p.type}: not logged in; run `{login}`"]
problems = []
if p.api_key_env and not os.environ.get(p.api_key_env):
problems.append(f"{p.name}: set {p.api_key_env}")
diff --git a/harness2/settings.py b/harness2/settings.py
index 5a1fcb1a..90f5d9a2 100644
--- a/harness2/settings.py
+++ b/harness2/settings.py
@@ -72,6 +72,8 @@
"base_url": "https://api.openai.com/v1",
"api_key_env": "OPENAI_API_KEY",
},
+ "claude-login": {"type": "claude-login"},
+ "chatgpt-login": {"type": "chatgpt-login"},
},
}
_MISSING = object()
diff --git a/harness2/substrate/claude.py b/harness2/substrate/claude.py
index c23b7f43..01f2574b 100644
--- a/harness2/substrate/claude.py
+++ b/harness2/substrate/claude.py
@@ -197,7 +197,15 @@ def strip_native(self, workspace: Path) -> None:
def install_into(self, install: Install, workspace: Path) -> None:
self.export_tree(self.import_tree(install.staging_dir), workspace, base={})
- def argv(self, prompt_file: Path, workspace: Path, model: str, effort: str) -> list[str]:
+ def argv(
+ self,
+ prompt_file: Path,
+ workspace: Path,
+ model: str,
+ effort: str,
+ *,
+ provider: providers.Provider | None = None,
+ ) -> list[str]:
return [
*cli_argv(self.bin, "claude"),
"-p",
@@ -211,16 +219,23 @@ def argv(self, prompt_file: Path, workspace: Path, model: str, effort: str) -> l
model,
"--effort",
effort,
+ *(
+ ["--strict-mcp-config", "--no-session-persistence"]
+ if provider is not None and provider.type == "claude-login"
+ else []
+ ),
]
def env(
self, base_env: Mapping[str, str], provider: providers.Provider, config_dir: Path
) -> dict[str, str]:
- config_dir.mkdir(parents=True, exist_ok=True)
+ login = provider.type == "claude-login"
+ if not login:
+ config_dir.mkdir(parents=True, exist_ok=True)
return {
**{k: v for k, v in base_env.items() if k not in providers.STRIP_ENV},
**providers.claude_env(provider),
- "CLAUDE_CONFIG_DIR": str(config_dir),
+ **({} if login else {"CLAUDE_CONFIG_DIR": str(config_dir)}),
}
def read_trajectory(
diff --git a/harness2/substrate/codex.py b/harness2/substrate/codex.py
index c555b230..2b8518fb 100644
--- a/harness2/substrate/codex.py
+++ b/harness2/substrate/codex.py
@@ -210,7 +210,15 @@ def strip_native(self, workspace: Path) -> None:
def install_into(self, install: Install, workspace: Path) -> None:
self.export_tree(self.import_tree(install.staging_dir), workspace, base={})
- def argv(self, prompt_file: Path, workspace: Path, model: str, effort: str) -> list[str]:
+ def argv(
+ self,
+ prompt_file: Path,
+ workspace: Path,
+ model: str,
+ effort: str,
+ *,
+ provider: providers.Provider | None = None,
+ ) -> list[str]:
return [
*cli_argv(self.bin, "codex"),
"exec",
@@ -219,6 +227,7 @@ def argv(self, prompt_file: Path, workspace: Path, model: str, effort: str) -> l
"--dangerously-bypass-approvals-and-sandbox",
"--dangerously-bypass-hook-trust",
"--ephemeral",
+ *providers.codex_args(provider, model, effort),
"-C",
str(workspace),
"-",
@@ -229,10 +238,19 @@ def write_home(self, home: Path, provider: providers.Provider, model: str, effor
home.mkdir(parents=True, exist_ok=True)
(home / "config.toml").write_text(content, encoding="utf-8")
- def env(self, base_env: Mapping[str, str], home: Path) -> dict[str, str]:
+ def env(
+ self,
+ base_env: Mapping[str, str],
+ home: Path | None,
+ *,
+ provider: providers.Provider | None = None,
+ ) -> dict[str, str]:
+ stripped = providers.STRIP_ENV
+ if provider is not None and provider.type == "chatgpt-login":
+ stripped += ("OPENAI_API_KEY", "CODEX_API_KEY")
return {
- **{k: v for k, v in base_env.items() if k not in providers.STRIP_ENV},
- "CODEX_HOME": str(home),
+ **{k: v for k, v in base_env.items() if k not in stripped},
+ **({"CODEX_HOME": str(home)} if home is not None else {}),
}
def final_message(self, stream_path: Path) -> str:
diff --git a/harness2/suite.py b/harness2/suite.py
index 2143948e..99af3eab 100644
--- a/harness2/suite.py
+++ b/harness2/suite.py
@@ -32,14 +32,14 @@
from pathlib import Path
from . import providers, workspace
-from .core import runmark, tree
+from .core import events, runmark, tree
from .core.library import slug
from .core.proposer import Proposer, ProposerError, _parse_json
from .core.spec import BenchPrompts, BenchSpec, ChannelSpec
from .registry import SuiteContext
from .settings import ConfigError, Settings
from .substrate._native import checked_path, cli_argv
-from .substrate.claude import ClaudeSubstrate
+from .substrate.claude import ClaudeSubstrate, parse_events
from .substrate.codex import CodexSubstrate
from .substrate.command import CommandSubstrate
@@ -450,10 +450,14 @@ def execute(
if self.executor["name"] == "claude":
env = self.substrate.env(env, provider, run_dir / "claude-home")
else:
- home = run_dir / "codex-home"
- self.substrate.write_home(home, provider, model, self.executor["effort"])
- env = self.substrate.env(env, home)
- argv = self.substrate.argv(prompt_file, ws, model, self.executor["effort"])
+ home = None
+ if provider.type != "chatgpt-login":
+ home = run_dir / "codex-home"
+ self.substrate.write_home(home, provider, model, self.executor["effort"])
+ env = self.substrate.env(env, home, provider=provider)
+ argv = self.substrate.argv(
+ prompt_file, ws, model, self.executor["effort"], provider=provider
+ )
try:
if self.executor["name"] == "command":
exit_code = self.substrate.run(prompt_file, ws, run_dir, wall, env)
@@ -477,6 +481,25 @@ def execute(
except OSError as exc:
(run_dir / "stderr.log").write_text(str(exc), encoding="utf-8")
exit_code = -1
+ stream = run_dir / "stream.jsonl"
+ failed = exit_code != 0 or any(
+ event.get("type") in ("error", "turn.failed")
+ or (
+ event.get("type") == "result"
+ and (event.get("is_error") or event.get("subtype") not in (None, "success"))
+ )
+ for event in parse_events(stream)
+ )
+ if failed:
+ tail = "".join(
+ path.read_text(encoding="utf-8", errors="replace")[-16000:]
+ for path in (stream, run_dir / "stderr.log")
+ if path.is_file()
+ )
+ if providers.USAGE_LIMIT.search(tail):
+ shutil.rmtree(run_dir)
+ events.request_stop("usage_limit")
+ raise events.Stopped("usage_limit")
if config["workspace"] == "project":
(run_dir / "deliverable.patch").write_text(
workspace.diff_since(ws, base, (*self.substrate.NATIVE_PATHS, ".harness2")),
@@ -484,7 +507,6 @@ def execute(
)
if (ws / "output").is_dir():
shutil.copytree(ws / "output", run_dir / "output")
- stream = run_dir / "stream.jsonl"
(run_dir / "final.md").write_text(
self.substrate.final_message(stream), encoding="utf-8"
)
diff --git a/harness2/workspace.py b/harness2/workspace.py
index 51b46414..e2de6a24 100644
--- a/harness2/workspace.py
+++ b/harness2/workspace.py
@@ -30,6 +30,13 @@
def project_workspace(project: Path, ref: str) -> Iterator[Path]:
ws = Path(tempfile.mkdtemp(prefix="harness2-ws-")).resolve()
try:
+ pruned = subprocess.run(
+ ["git", "-C", str(project), "worktree", "prune"],
+ capture_output=True,
+ text=True,
+ )
+ if pruned.returncode:
+ raise ConfigError(f"cannot prune project worktrees: {pruned.stderr.strip()}")
result = subprocess.run(
["git", "-C", str(project), "worktree", "add", "--detach", "--", str(ws), ref],
capture_output=True,
diff --git a/tests/test_providers.py b/tests/test_providers.py
index 96c5efc7..a58f67f7 100644
--- a/tests/test_providers.py
+++ b/tests/test_providers.py
@@ -43,7 +43,10 @@ def test_provider_is_frozen_and_strip_env_is_exact(self):
provider = Provider("custom", "anthropic")
with self.assertRaises(FrozenInstanceError):
provider.region = "changed"
- self.assertEqual(providers.PROVIDER_TYPES, ("anthropic", "openai", "vertex", "bedrock"))
+ self.assertEqual(
+ providers.PROVIDER_TYPES,
+ ("anthropic", "openai", "vertex", "bedrock", "claude-login", "chatgpt-login"),
+ )
self.assertEqual(
providers.STRIP_ENV,
(
diff --git a/tests/test_subscriptions.py b/tests/test_subscriptions.py
new file mode 100644
index 00000000..4adfcc1f
--- /dev/null
+++ b/tests/test_subscriptions.py
@@ -0,0 +1,643 @@
+# Copyright 2026 The Harness2 Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Subscription isolation and resumable usage limits against offline CLIs."""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import tomllib
+import unittest
+from pathlib import Path
+from unittest.mock import Mock, patch
+
+from harness2 import providers, settings, suite, workspace
+from harness2.cli.main import main
+from harness2.core import events, recursion, runmark, schedule
+from harness2.core.proposer import Proposer
+from harness2.core.spec import ChannelSpec
+from harness2.registry import SuiteContext
+from harness2.settings import ConfigError
+
+FAKE_CLI = r"""import json, os, sys
+from pathlib import Path
+
+argv = sys.argv[1:]
+kind = Path(sys.argv[0]).name
+if "--version" in argv:
+ print("fixture 1.0")
+ sys.exit(0)
+status = argv in (["login", "status"], ["auth", "status"])
+cwd = Path.cwd()
+home = Path(os.environ.get("CODEX_HOME", Path.home() / ".codex"))
+config = home / "config.toml"
+record = {"kind": kind, "argv": argv, "env": dict(os.environ), "cwd": str(cwd),
+ "stdin": "" if status else sys.stdin.read(),
+ "config_exists": config.exists(),
+ "config": config.read_text() if config.exists() else None}
+with open(os.environ["FAKE_RECORD"], "a") as log:
+ log.write(json.dumps(record) + "\n")
+if status:
+ sys.exit(int(os.environ.get("FAKE_LOGIN_RC", "0")))
+improver = ("--output-format" in argv and argv[argv.index("--output-format") + 1] == "json"
+ or (cwd / "context").is_dir())
+role = "improver" if improver else "executor"
+if os.environ.get("LIMIT_ROLE") == role:
+ limit = os.environ.get("LIMIT_TEXT", "rate-limit reached")
+ source = os.environ.get("LIMIT_SOURCE", "event")
+ if source == "event":
+ if kind == "claude":
+ print(json.dumps({"type": "result", "subtype": "error_during_execution",
+ "is_error": True, "errors": [limit]}))
+ else:
+ print(json.dumps({"type": "turn.failed", "error": {"message": limit}}))
+ sys.exit(0)
+ print(limit, file=sys.stderr if source == "stderr" else sys.stdout)
+ sys.exit(7)
+if improver:
+ context = cwd / "context"
+ if (context / "candidates").is_dir():
+ chosen = sorted((context / "candidates").glob("*.md"))[0].stem
+ text = "Complete.\n```choice\n" + chosen + "\n```"
+ elif (context / "adoption.json").is_file():
+ text = "NO_ENTRY"
+ else:
+ for name in ("CLAUDE.md", "AGENTS.md"):
+ harness = context / "harness" / name
+ if harness.is_file():
+ harness.write_text(harness.read_text() + "Check the output.\n")
+ text = "Check the output."
+else:
+ (cwd / "output").mkdir(exist_ok=True)
+ (cwd / "output/report.txt").write_text("done\n")
+ text = "done"
+if kind == "claude":
+ print(json.dumps({"type": "result", "subtype": "success", "result": text,
+ "total_cost_usd": 0.01}))
+else:
+ print(json.dumps({"type": "item.completed", "item": {
+ "id": "reply", "type": "agent_message", "text": text}}))
+"""
+
+
+class SubscriptionTests(unittest.TestCase):
+ maxDiff = None
+
+ def setUp(self):
+ temporary = tempfile.TemporaryDirectory()
+ self.addCleanup(temporary.cleanup)
+ self.root = Path(temporary.name).resolve()
+ self.project = self.root / "project"
+ self.tasks = self.project / ".harness2/tasks"
+ self.tasks.mkdir(parents=True)
+ (self.project / "CLAUDE.md").write_text("Read the task.\n")
+ (self.project / "AGENTS.md").write_text("Read the task.\n")
+ binary = self.root / "bin"
+ binary.mkdir()
+ for name in ("claude", "codex"):
+ path = binary / name
+ path.write_text(f"#!{sys.executable}\n" + FAKE_CLI)
+ path.chmod(0o755)
+ self.record = self.root / "records.jsonl"
+ self.user_home = self.root / "user"
+ for name in (".codex", ".claude"):
+ path = self.user_home / name
+ path.mkdir(parents=True)
+ (path / "config.toml").write_text("private user config")
+ (path / "auth.json").write_text("private user auth")
+ environment = patch.dict(
+ os.environ,
+ {
+ "PATH": f"{binary}{os.pathsep}{os.defpath}",
+ "HOME": str(self.user_home),
+ "XDG_CONFIG_HOME": str(self.root / "config"),
+ "FAKE_RECORD": str(self.record),
+ "CLAUDE_CODE_OAUTH_TOKEN": "fixture-oauth",
+ "ANTHROPIC_API_KEY": "ambient-anthropic",
+ "OPENAI_API_KEY": "ambient-openai",
+ "CODEX_API_KEY": "ambient-codex",
+ "ANTHROPIC_BASE_URL": "https://ambient.invalid",
+ "OPENAI_BASE_URL": "https://ambient.invalid",
+ "PYTHONDONTWRITEBYTECODE": "1",
+ },
+ clear=True,
+ )
+ environment.start()
+ self.addCleanup(environment.stop)
+ events.start(None)
+ self.addCleanup(events.start, None)
+ task = suite.new_task(self.tasks, "one", files=True, domain="coding", check="true")
+ (task / "task.md").write_text("Produce a report.")
+
+ def adapter(self, name="test", **overrides):
+ values = {
+ "executor.model": "claude-login/claude-sonnet-5-5",
+ "improver.model": "claude-login/claude-sonnet-5-5",
+ }
+ values.update(overrides)
+ config = settings.load(
+ self.project, overrides={key: (value, "test") for key, value in values.items()}
+ )
+ return suite.make_adapter(
+ SuiteContext(config, self.project, self.project / ".harness2/runs" / name)
+ )
+
+ def records(self):
+ return [json.loads(line) for line in self.record.read_text().splitlines()]
+
+ def launch_record(self, label):
+ record = self.records()[-1]
+ env = dict(record["env"])
+ env.pop("__CF_USER_TEXT_ENCODING", None)
+ normalized = json.dumps(
+ {"argv": [record["kind"], *record["argv"]], "env": env}, sort_keys=True
+ )
+ normalized = normalized.replace(record["cwd"], "")
+ normalized = normalized.replace(str(self.root), "")
+ print(f"SUBSCRIPTION {label}: {normalized}")
+ return record
+
+ def test_builtin_providers_use_cli_auth(self):
+ config = settings.load(self.project)
+ for name in ("claude-login", "chatgpt-login"):
+ self.assertEqual(config.provider(name).type, name)
+ self.assertEqual(providers.claude_env(config.provider("claude-login")), {})
+ with patch.dict(os.environ, {}, clear=True):
+ self.assertEqual(providers.claude_env(config.provider("claude-login")), {})
+
+ def test_claude_subscription_executor_and_improver_launches(self):
+ adapter = self.adapter()
+ result = adapter.execute("one", None, "claude")
+ record = self.launch_record("claude-login executor")
+ self.assertEqual(
+ record["argv"],
+ [
+ "-p",
+ "--output-format",
+ "stream-json",
+ "--verbose",
+ "--dangerously-skip-permissions",
+ "--setting-sources",
+ "project,local",
+ "--model",
+ "claude-sonnet-5-5",
+ "--effort",
+ "high",
+ "--strict-mcp-config",
+ "--no-session-persistence",
+ ],
+ )
+ self.assertFalse((result / "claude-home").exists())
+ context = self.root / "context"
+ context.mkdir()
+ Proposer(adapter.spec.channels["proposer"]).call("mission", context, "improver")
+ record = self.launch_record("claude-login improver")
+ self.assertEqual(
+ record["argv"],
+ [
+ "--dangerously-skip-permissions",
+ "-p",
+ "--output-format",
+ "json",
+ "--effort",
+ "high",
+ "--model",
+ "claude-sonnet-5-5",
+ "--setting-sources",
+ "project,local",
+ "--strict-mcp-config",
+ "--no-session-persistence",
+ ],
+ )
+ for record in self.records():
+ self.assertNotIn("CLAUDE_CONFIG_DIR", record["env"])
+ self.assertEqual(record["env"]["CLAUDE_CODE_OAUTH_TOKEN"], "fixture-oauth")
+ self.assertTrue(all(key not in record["env"] for key in providers.STRIP_ENV))
+ os.environ.pop("CLAUDE_CODE_OAUTH_TOKEN")
+ adapter.execute("one", None, "native-login")
+ self.assertNotIn("CLAUDE_CODE_OAUTH_TOKEN", self.records()[-1]["env"])
+ self.assertNotIn("CLAUDE_CONFIG_DIR", self.records()[-1]["env"])
+
+ def test_chatgpt_subscription_executor_and_improver_launches(self):
+ adapter = self.adapter(
+ **{
+ "executor.name": "codex",
+ "executor.model": "chatgpt-login/gpt-6.1-sol",
+ "improver.channel": "codex",
+ "improver.model": "chatgpt-login/gpt-6.1-sol",
+ }
+ )
+ home = self.user_home / ".codex"
+ before = {p.name: p.read_bytes() for p in home.iterdir()}
+ result = adapter.execute("one", None, "codex")
+ record = self.launch_record("chatgpt-login executor")
+ self.assertEqual(
+ record["argv"],
+ [
+ "exec",
+ "--json",
+ "--skip-git-repo-check",
+ "--dangerously-bypass-approvals-and-sandbox",
+ "--dangerously-bypass-hook-trust",
+ "--ephemeral",
+ "--ignore-user-config",
+ "--ignore-rules",
+ "-m",
+ "gpt-6.1-sol",
+ "-c",
+ 'model_reasoning_effort="high"',
+ "-C",
+ record["cwd"],
+ "-",
+ ],
+ )
+ self.assertFalse((result / "codex-home").exists())
+ context = self.root / "context"
+ context.mkdir()
+ Proposer(adapter.spec.channels["proposer"]).call("mission", context, "improver")
+ record = self.launch_record("chatgpt-login improver")
+ self.assertEqual(
+ record["argv"],
+ [
+ "exec",
+ "--json",
+ "--skip-git-repo-check",
+ "--dangerously-bypass-approvals-and-sandbox",
+ "--ephemeral",
+ "--ignore-user-config",
+ "--ignore-rules",
+ "-m",
+ "gpt-6.1-sol",
+ "-c",
+ 'model_reasoning_effort="high"',
+ "-C",
+ record["cwd"],
+ "-",
+ ],
+ )
+ for record in self.records():
+ self.assertNotIn("CODEX_HOME", record["env"])
+ self.assertTrue(
+ all(
+ key not in record["env"]
+ for key in (*providers.STRIP_ENV, "OPENAI_API_KEY", "CODEX_API_KEY")
+ )
+ )
+ self.assertEqual({p.name: p.read_bytes() for p in home.iterdir()}, before)
+
+ def test_inherited_login_homes_are_preserved(self):
+ context = self.root / "context"
+ context.mkdir()
+ for kind, provider_name, variable in (
+ ("claude", "claude-login", "CLAUDE_CONFIG_DIR"),
+ ("codex", "chatgpt-login", "CODEX_HOME"),
+ ):
+ with (
+ self.subTest(kind=kind),
+ patch.dict(os.environ, {variable: str(self.user_home / ("." + kind))}),
+ ):
+ adapter = self.adapter(
+ kind,
+ **{
+ "executor.name": kind,
+ "executor.model": provider_name + "/model",
+ "improver.channel": kind,
+ "improver.model": provider_name + "/model",
+ },
+ )
+ adapter.execute("one", None, "inherited")
+ self.assertEqual(self.records()[-1]["env"][variable], os.environ[variable])
+ Proposer(adapter.spec.channels["proposer"]).call("mission", context, "inherited")
+ self.assertEqual(self.records()[-1]["env"][variable], os.environ[variable])
+
+ def test_login_status_success_failure_missing_cli_and_timeout(self):
+ for kind, argv, login in (
+ ("claude-login", ["auth", "status"], "claude auth login"),
+ ("chatgpt-login", ["login", "status"], "codex login"),
+ ):
+ with self.subTest(provider=kind):
+ provider = providers.Provider(kind, kind)
+ self.assertEqual(providers.check(provider), [])
+ record = self.records()[-1]
+ self.assertEqual(record["argv"], argv)
+ self.assertNotIn("CLAUDE_CONFIG_DIR", record["env"])
+ self.assertNotIn("CODEX_HOME", record["env"])
+ self.assertTrue(all(key not in record["env"] for key in providers.STRIP_ENV))
+ problem = [f"{kind}: not logged in; run `{login}`"]
+ with patch.dict(os.environ, {"FAKE_LOGIN_RC": "1"}):
+ self.assertEqual(providers.check(provider), problem)
+ with patch.dict(os.environ, {"PATH": ""}):
+ self.assertEqual(providers.check(provider), problem)
+ with patch(
+ "harness2.providers.subprocess.run",
+ side_effect=subprocess.TimeoutExpired("cli", 10),
+ ):
+ self.assertEqual(providers.check(provider), problem)
+ args = providers.codex_args(
+ providers.Provider("chatgpt-login", "chatgpt-login"),
+ 'model"\nnext\x7f',
+ 'effort"\nnext',
+ )
+ self.assertEqual(args[3], 'model"\nnext\x7f')
+ self.assertEqual(tomllib.loads(args[5]), {"model_reasoning_effort": 'effort"\nnext'})
+
+ def test_api_provider_homes_and_ambient_oauth_are_unchanged(self):
+ for kind, provider_name in (("claude", "anthropic"), ("codex", "openai")):
+ with self.subTest(kind=kind):
+ adapter = self.adapter(
+ kind,
+ **{
+ "executor.name": kind,
+ "executor.model": provider_name + "/fixture",
+ "improver.channel": kind,
+ "improver.model": provider_name + "/fixture",
+ },
+ )
+ result = adapter.execute("one", None, "api")
+ record = self.records()[-1]
+ self.assertEqual(record["env"]["CLAUDE_CODE_OAUTH_TOKEN"], "fixture-oauth")
+ if kind == "claude":
+ self.assertEqual(record["env"]["ANTHROPIC_API_KEY"], "ambient-anthropic")
+ else:
+ self.assertEqual(record["env"]["OPENAI_API_KEY"], "ambient-openai")
+ self.assertEqual(record["env"]["CODEX_HOME"], str(result / "codex-home"))
+ self.assertEqual(
+ record["config"],
+ providers.codex_config(
+ adapter.settings.provider("openai"), "fixture", "high"
+ ),
+ )
+ self.assertNotIn("-c", record["argv"])
+ Proposer(adapter.spec.channels["proposer"]).call("mission", None, "api-improver")
+ record = self.records()[-1]
+ self.assertEqual(record["env"]["CLAUDE_CODE_OAUTH_TOKEN"], "fixture-oauth")
+ if kind == "codex":
+ self.assertEqual(
+ record["config"],
+ providers.codex_config(
+ adapter.settings.provider("openai"), "fixture", "high"
+ ),
+ )
+ self.assertNotEqual(record["env"]["CODEX_HOME"], str(self.user_home / ".codex"))
+
+ def test_wrong_cli_and_bridge_reject_subscription_providers(self):
+ claude = providers.Provider("claude-login", "claude-login")
+ chatgpt = providers.Provider("chatgpt-login", "chatgpt-login")
+ with self.assertRaisesRegex(ConfigError, "cannot use a ChatGPT login"):
+ providers.claude_env(chatgpt)
+ with self.assertRaises(ConfigError):
+ providers.codex_config(claude, "model", "high")
+ with self.assertRaises(ConfigError):
+ providers.codex_args(claude, "model", "high")
+ for provider in (claude, chatgpt):
+ for operation in (providers.opencode_config, providers.litellm_params):
+ with self.subTest(provider=provider.type, operation=operation.__name__):
+ with self.assertRaisesRegex(ConfigError, "API provider"):
+ operation(provider, "model")
+
+ def test_doctor_per_role_subscription_compatibility(self):
+ for channel, provider_name in (
+ ("claude", "claude-login"),
+ ("codex", "chatgpt-login"),
+ ("claude", "chatgpt-login"),
+ ("codex", "claude-login"),
+ ("opencode", "claude-login"),
+ ("opencode", "chatgpt-login"),
+ ):
+ with self.subTest(channel=channel, provider=provider_name):
+ adapter = (
+ self.adapter(
+ **{
+ "executor.name": channel,
+ "executor.model": provider_name + "/model",
+ "improver.channel": channel,
+ "improver.model": provider_name + "/model",
+ }
+ )
+ if channel != "opencode"
+ else self.adapter(
+ **{
+ "improver.channel": channel,
+ "improver.model": provider_name + "/model",
+ }
+ )
+ )
+ code, roles = self.doctor(adapter)
+ compatible = (channel, provider_name) in (
+ ("claude", "claude-login"),
+ ("codex", "chatgpt-login"),
+ )
+ self.assertEqual(code, 0 if compatible else 1)
+ self.assertEqual(roles["improver"]["ok"], compatible)
+ self.assertEqual(roles["judge"]["ok"], compatible)
+
+ def doctor(self, adapter):
+ output = io.StringIO()
+ argv = ["-C", str(self.project), "--json"]
+ for key, value in adapter.settings.flat().items():
+ if key.startswith(("executor.", "improver.")) and not isinstance(value, list):
+ argv += ["-c", f"{key}={json.dumps(value)}"]
+ with contextlib.redirect_stdout(output):
+ code = main([*argv, "doctor"])
+ return code, {row["check"]: row for row in json.loads(output.getvalue())}
+
+ def test_doctor_login_status_and_global_codex_instructions(self):
+ adapter = self.adapter(
+ **{
+ "executor.name": "codex",
+ "executor.model": "chatgpt-login/model",
+ "improver.channel": "codex",
+ "improver.model": "chatgpt-login/model",
+ }
+ )
+ agents = self.user_home / ".codex/AGENTS.md"
+ agents.write_text("Global rules. 한글\n")
+ code, rows = self.doctor(adapter)
+ self.assertEqual(code, 0)
+ self.assertEqual(
+ rows["Codex global AGENTS.md"],
+ {
+ "check": "Codex global AGENTS.md",
+ "ok": True,
+ "detail": "your global AGENTS.md (17 characters) is part of every Codex run",
+ },
+ )
+ with patch.dict(os.environ, {"FAKE_LOGIN_RC": "1"}):
+ code, rows = self.doctor(adapter)
+ self.assertEqual(code, 1)
+ self.assertEqual(
+ rows["provider chatgpt-login"]["detail"],
+ "chatgpt-login: not logged in; run `codex login`",
+ )
+ code, rows = self.doctor(self.adapter())
+ self.assertEqual(code, 1)
+ self.assertEqual(
+ rows["provider claude-login"]["detail"],
+ "claude-login: not logged in; run `claude auth login`",
+ )
+ _, rows = self.doctor(self.adapter())
+ self.assertNotIn("Codex global AGENTS.md", rows)
+ agents.write_text("")
+ _, rows = self.doctor(adapter)
+ self.assertNotIn("Codex global AGENTS.md", rows)
+ other = self.root / "custom-codex"
+ other.mkdir()
+ (other / "AGENTS.md").write_text("custom")
+ with patch.dict(os.environ, {"CODEX_HOME": str(other)}):
+ _, rows = self.doctor(adapter)
+ self.assertEqual(
+ rows["Codex global AGENTS.md"]["detail"],
+ "your global AGENTS.md (6 characters) is part of every Codex run",
+ )
+ self.assertTrue(
+ all(
+ record["argv"] in (["auth", "status"], ["login", "status"])
+ for record in self.records()
+ )
+ )
+
+ def test_executor_usage_limits_stop_and_resume_entire_run(self):
+ adapter = self.adapter()
+ self.stop_and_resume(adapter, "executor")
+
+ def test_improver_usage_limits_stop_without_retry_and_resume_entire_run(self):
+ adapter = self.adapter()
+ self.stop_and_resume(adapter, "improver")
+
+ def stop_and_resume(self, adapter, role):
+ run_root = adapter.spec.runs_root
+ event_path = run_root / "events.jsonl"
+ prop = Proposer(adapter.spec.channels["proposer"], audit_dir=run_root / "audit")
+ kwargs = dict(
+ adapter=adapter,
+ prop=prop,
+ runs_root=run_root,
+ library_root=run_root / "library",
+ modes=("agg",),
+ )
+ events.start(event_path)
+ with (
+ patch.dict(os.environ, {"LIMIT_ROLE": role}),
+ patch("harness2.core.proposer.time", wraps=time) as clock,
+ ):
+ with self.assertRaisesRegex(events.Stopped, "^usage_limit$"):
+ schedule.run_sweep(1, **kwargs)
+ clock.sleep.assert_not_called()
+ self.assertEqual(events.totals()["stopped"], "usage_limit")
+ self.assertFalse(list(run_root.glob("tasks-*/**/summary.json")))
+ if role == "executor":
+ self.assertEqual(list((run_root / "rollouts").iterdir()), [])
+ self.assertEqual(events.totals()["calls"], 0)
+ else:
+ self.assertFalse(list((run_root / "audit").glob("*.response.md")))
+ events.start(event_path)
+ resumed = schedule.run_sweep(1, **kwargs)
+ self.assertEqual((resumed["domains"][0]["done"], resumed["domains"][0]["failed"]), (1, 0))
+ self.assertTrue(list(run_root.glob("tasks-*/**/summary.json")))
+ rollouts = list((run_root / "rollouts").iterdir())
+ self.assertTrue(rollouts)
+ self.assertTrue(all(runmark.is_complete(path, match_harness=False) for path in rollouts))
+
+ def test_usage_limit_failure_sources_and_success_mentions(self):
+ for source in ("event", "stderr", "stdout"):
+ for kind in ("claude", "codex"):
+ with self.subTest(source=source, kind=kind):
+ provider_name = "claude-login" if kind == "claude" else "chatgpt-login"
+ adapter = self.adapter(
+ f"{kind}-{source}",
+ **{
+ "executor.name": kind,
+ "executor.model": provider_name + "/model",
+ },
+ )
+ events.start(None)
+ with patch.dict(os.environ, {"LIMIT_ROLE": "executor", "LIMIT_SOURCE": source}):
+ with self.assertRaisesRegex(events.Stopped, "usage_limit"):
+ adapter.execute("one", None, "limited")
+ self.assertEqual(list((adapter.spec.runs_root / "rollouts").iterdir()), [])
+ events.start(None)
+ provider = adapter.settings.provider(provider_name)
+ with patch.dict(os.environ, {"LIMIT_ROLE": "improver", "LIMIT_SOURCE": source}):
+ context = self.root / "context"
+ context.mkdir(exist_ok=True)
+ with self.assertRaisesRegex(events.Stopped, "usage_limit"):
+ Proposer(
+ ChannelSpec(kind, provider_name + "/model", provider=provider)
+ ).call("mission", context, "limited")
+ for text in (
+ "usage limit",
+ "RATE_LIMIT",
+ "too many requests",
+ "HTTP 429",
+ "quota",
+ "credit balance is too low",
+ ):
+ self.assertIsNotNone(providers.USAGE_LIMIT.search(text))
+ self.assertIsNone(providers.USAGE_LIMIT.search("14290"))
+ events.start(None)
+ adapter = self.adapter("success")
+ (self.root / "bin/claude").write_text(
+ f"#!{sys.executable}\nimport json\n"
+ "print(json.dumps({'type':'result', 'subtype':'success', 'result':'rate limit fixed'}))\n"
+ )
+ self.assertTrue(
+ runmark.is_complete(
+ adapter.execute("one", None, "success"), harness_dir=adapter.base_harness()
+ )
+ )
+ self.assertTrue(
+ Proposer(adapter.spec.channels["proposer"]).call("mission", None, "success")
+ )
+
+ def test_core_does_not_swallow_stopped_from_batched_adapter(self):
+ adapter = self.adapter()
+ adapter.execute_many = Mock(side_effect=events.Stopped("usage_limit"))
+ with self.assertRaisesRegex(events.Stopped, "usage_limit"):
+ schedule._roll_bases("one", adapter, [("a", "a"), ("b", "b")], self.root / "dead")
+ with patch.object(adapter, "execute", side_effect=events.Stopped("usage_limit")):
+ with self.assertRaisesRegex(events.Stopped, "usage_limit"):
+ recursion._roll(adapter, None, "one", "a", None, lambda label: label, self.root)
+
+ def test_opencode_usage_limit_stops_even_with_a_partial_answer(self):
+ binary = self.root / "bin/opencode"
+ binary.write_text(
+ f"#!{sys.executable}\nimport json\nfrom pathlib import Path\n"
+ "Path('output/answer.md').write_text('partial')\n"
+ "print(json.dumps({'type':'error','error':{'data':{'message':'quota exceeded'}}}))\n"
+ )
+ binary.chmod(0o755)
+ provider = providers.Provider("api", "openai")
+ prop = Proposer(ChannelSpec("opencode", "api/model", provider=provider))
+ with patch("harness2.core.proposer.time", wraps=time) as clock:
+ with self.assertRaisesRegex(events.Stopped, "usage_limit"):
+ prop.call("mission", None, "limit")
+ clock.sleep.assert_not_called()
+ self.assertEqual(events.totals()["stopped"], "usage_limit")
+
+ def test_project_workspace_prunes_before_adding(self):
+ subprocess.run(["git", "init", "-q", str(self.project)], check=True)
+ workspace.commit_all(self.project, "fixture")
+ with patch("harness2.workspace.subprocess.run", wraps=subprocess.run) as run:
+ with workspace.project_workspace(self.project, "HEAD") as ws:
+ self.assertTrue(ws.exists())
+ argv = [call.args[0] for call in run.call_args_list]
+ self.assertEqual(argv[0], ["git", "-C", str(self.project), "worktree", "prune"])
+ self.assertEqual(argv[1][4], "add")