From 0d1b75d738547e749273be9a5e952882f0b3c8ad Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Thu, 29 Oct 2026 10:00:38 -0600 Subject: [PATCH 1/8] feat(harness): experimental agentic testing proof of concept (untested) Theorized proof of concept for exercising the agent runtime from the harness. Not yet run against a real server; committed so we can test how this behaves in reality and iterate. Agentic testing is off by default (HARNESS_AGENTIC_TESTING_ENABLED), so the existing deployment behaves as before. - main.py: split into modules with prepare/start/stop/wait so every module forks its worker processes before any governor thread starts - sdk_testing.py: the existing simulated-task testing, moved unchanged - harness_env.py: env helpers shared by both modules - agentic/: a storyteller agent writes one sentence per turn and calls a critic sub-agent; its stop_when worker ends the loop when the critic's JSON verdict says the story meets a randomly invented concept, or max_turns ends it. A governor starts runs on an interval and swaps in a new concept every few runs by redeploying under the same name. - try_agentic.py: runs one plan or one story locally without the governors, cancelling the run on Ctrl-C or timeout Ideas to test: - whether the critic's verdict reaches stop_when through the LLM's message history, and what shape it arrives in - whether redeploying under the same name replaces version 1 in place - behaviour on Orkes 5.6.0-rc4 (OSS 3.32.1) vs OSS 3.33.0 - record/replay through mock/mockLLM with a fixed seed (OSS 3.33.0+) - real token cost per story and how many turns concepts take Co-Authored-By: Claude Opus 5.5 --- harness/README.md | 68 +++++++ harness/agentic/__init__.py | 3 + harness/agentic/agentic_governor.py | 82 +++++++++ harness/agentic/agentic_testing.py | 62 +++++++ harness/agentic/story_agents.py | 152 ++++++++++++++++ harness/agentic/story_concepts.py | 95 ++++++++++ harness/harness_env.py | 29 +++ harness/main.py | 129 ++++---------- harness/sdk_testing.py | 122 +++++++++++++ harness/try_agentic.py | 265 ++++++++++++++++++++++++++++ 10 files changed, 913 insertions(+), 94 deletions(-) create mode 100644 harness/agentic/__init__.py create mode 100644 harness/agentic/agentic_governor.py create mode 100644 harness/agentic/agentic_testing.py create mode 100644 harness/agentic/story_agents.py create mode 100644 harness/agentic/story_concepts.py create mode 100644 harness/harness_env.py create mode 100644 harness/sdk_testing.py create mode 100644 harness/try_agentic.py diff --git a/harness/README.md b/harness/README.md index 01346f4bb..e5cbc0a63 100644 --- a/harness/README.md +++ b/harness/README.md @@ -99,3 +99,71 @@ All resource names use a `python_` prefix so multiple SDK harnesses (C#, JS, Go, | `HARNESS_WORKFLOWS_PER_SEC` | no | 2 | Workflows to start per second | | `HARNESS_BATCH_SIZE` | no | 20 | Number of tasks each worker polls per batch | | `HARNESS_POLL_INTERVAL_MS` | no | 100 | Milliseconds between poll cycles | +| `HARNESS_SDK_TESTING_ENABLED` | no | true | Run the simulated-task workflows described above | +| `HARNESS_AGENTIC_TESTING_ENABLED` | no | false | Run the agentic storyteller described below | + +## Agentic Testing + +Off by default. When enabled, the harness also exercises the agent runtime. It compiles a `python_storyteller` agent from a randomly chosen story concept (protagonist, conflict, resolution, constraint, tone), serves the agent's `python_storyteller_stop_when` worker from this process, and starts a run on an interval, each in a randomly chosen setting. + +Each run adds one sentence per turn. After every sentence the storyteller calls a `python_story_critic` sub-agent, which returns a JSON verdict: whether the story meets the concept, and what is still missing. The `stop_when` worker ends the loop once the critic approves. If the critic never approves, `HARNESS_AGENTIC_MAX_TURNS` ends it. Either way the run completes. After `HARNESS_AGENTIC_RUNS_PER_CONCEPT` runs, the harness invents a new concept and redeploys over the same name, which replaces version 1 in place rather than adding versions. + +The worker logs each verdict it sees: + +```text +[python_storyteller] turn 3: latest verdict done=False missing=['introduce the conflict'] +[python_storyteller] turn 7: latest verdict done=True missing=[] +``` + +### Requirements + +The LLM call runs on the Conductor server, so the provider key belongs in the **server's** environment, not the harness's. Conductor OSS enables the AI integration by default and reads `ANTHROPIC_API_KEY` into `conductor.ai.anthropic.api-key`. + +### Running Locally Against Conductor OSS + +```bash +# Terminal 1: start the server with the provider key in its environment +export ANTHROPIC_API_KEY=... +npm install -g @conductor-oss/conductor-cli +conductor server start + +# Terminal 2: agentic testing only +export CONDUCTOR_SERVER_URL=http://localhost:8080/api +HARNESS_SDK_TESTING_ENABLED=false \ +HARNESS_AGENTIC_TESTING_ENABLED=true \ +HARNESS_AGENTIC_INTERVAL_SECONDS=60 \ + python3 harness/main.py +``` + +Open the execution in the Conductor UI to see each storyteller turn and critic call. + +### Trying One Run + +`try_agentic.py` runs the storyteller once, without the governors or the simulated-task workflows. It reads the same `CONDUCTOR_SERVER_URL`, `CONDUCTOR_AUTH_KEY` and `CONDUCTOR_AUTH_SECRET` variables; everything else is a flag. + +```bash +# Ask the server to compile the agent. Nothing runs and no LLM is called. +# Confirms the agent API is on and shows the compiled workflow. +python3 harness/try_agentic.py plan + +# Run one story: one line per sentence and critic verdict, then a report. +python3 harness/try_agentic.py run + +# Repeat a concept and setting, with a shorter limit +python3 harness/try_agentic.py run --seed 3f9a1c2e --setting "a flooded library" --max-turns 10 +``` + +A run is cancelled on Ctrl-C or after `--timeout` seconds (default 600), so it doesn't keep calling the LLM. It registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. On Orkes, `--model` takes the AI integration's name before the slash. + +### Agentic Environment Variables + +| Variable | Default | Description | +|---|---|---| +| `HARNESS_AGENTIC_MODEL` | `anthropic/claude-haiku-4-5` | Model for the storyteller and the critic, in `provider/model` form | +| `HARNESS_AGENTIC_INTERVAL_SECONDS` | 300 | Seconds between run starts | +| `HARNESS_AGENTIC_RUNS_PER_CONCEPT` | 3 | Runs per concept before a new one is invented and redeployed | +| `HARNESS_AGENTIC_MAX_TURNS` | 20 | Turn limit for a run whose story the critic never approves | +| `HARNESS_AGENTIC_SEED` | random | Fixed seed for a reproducible concept and setting sequence. Required to replay recordings through `mock/mockLLM` | +| `HARNESS_AGENTIC_DEBUG` | false | Print the raw message history when the worker finds no verdict | + +**Cost:** each run makes two LLM calls per turn, one for the storyteller and one for the critic. Because every turn resends the growing story, input tokens grow faster than linearly with turn count. On Claude Haiku 4.5 a 15-turn run costs roughly $0.04. At the default 300-second interval that is about $10 per day. Lower the cadence before you leave it running. diff --git a/harness/agentic/__init__.py b/harness/agentic/__init__.py new file mode 100644 index 000000000..39f21009f --- /dev/null +++ b/harness/agentic/__init__.py @@ -0,0 +1,3 @@ +from agentic.agentic_testing import AgenticTesting + +__all__ = ["AgenticTesting"] diff --git a/harness/agentic/agentic_governor.py b/harness/agentic/agentic_governor.py new file mode 100644 index 000000000..378a4024f --- /dev/null +++ b/harness/agentic/agentic_governor.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import random +import threading + +from conductor.ai.agents import AgentRuntime + +from agentic.story_agents import STORYTELLER_NAME, build_storyteller, story_prompt +from agentic.story_concepts import StoryConcept, invent_concept, pick_setting + + +class AgenticGovernor: + """Starts one storyteller run per interval, each in a randomly picked setting. + + After ``runs_per_concept`` runs it invents a new concept and redeploys the + storyteller under the same name. The server upserts agent definitions on + (name, version) and the compiler always emits version 1, so a redeploy + overwrites the previous concept in place -- definitions never accumulate. + Redeploying is plain HTTP and registers no workers, so it is safe from this + thread; starting by name likewise skips worker registration. + """ + + def __init__( + self, + runtime: AgentRuntime, + rng: random.Random, + concept: StoryConcept, + model: str, + max_turns: int, + interval_seconds: int, + runs_per_concept: int, + ) -> None: + self._runtime = runtime + self._rng = rng + self._concept = concept + self._model = model + self._max_turns = max_turns + self._interval_seconds = interval_seconds + self._runs_per_concept = runs_per_concept + self._runs_left = runs_per_concept + self._stop_event = threading.Event() + self._thread: threading.Thread | None = None + + def start(self) -> None: + print( + f"AgenticGovernor started: agent={STORYTELLER_NAME}, " + f"interval={self._interval_seconds}s, runs_per_concept={self._runs_per_concept}" + ) + self._stop_event.clear() + self._thread = threading.Thread(target=self._run, name="AgenticGovernor", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + if self._thread is not None: + self._thread.join(timeout=5.0) + print("AgenticGovernor stopped") + + def _run(self) -> None: + while not self._stop_event.is_set(): + self._start_run() + self._stop_event.wait(timeout=self._interval_seconds) + + def _start_run(self) -> None: + try: + if self._runs_left <= 0: + self._rotate_concept() + setting = pick_setting(self._rng) + handle = self._runtime.start( + STORYTELLER_NAME, story_prompt(setting), version=1, + ) + self._runs_left -= 1 + print(f"AgenticGovernor: started {handle.execution_id} (setting: {setting})") + except Exception as e: + print(f"AgenticGovernor: error starting storyteller run: {e}") + + def _rotate_concept(self) -> None: + concept = invent_concept(self._rng) + self._runtime.deploy(build_storyteller(concept, self._model, self._max_turns)) + self._concept = concept + self._runs_left = self._runs_per_concept + print(f"AgenticGovernor: deployed new concept: {concept.summary()}") diff --git a/harness/agentic/agentic_testing.py b/harness/agentic/agentic_testing.py new file mode 100644 index 000000000..24e8bea5f --- /dev/null +++ b/harness/agentic/agentic_testing.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import os +import random +import threading + +from conductor.ai.agents import AgentRuntime +from conductor.client.configuration.configuration import Configuration + +from agentic.agentic_governor import AgenticGovernor +from agentic.story_agents import STORYTELLER_NAME, build_storyteller +from agentic.story_concepts import invent_concept +from harness_env import env_int_or_default, env_str_or_default + + +class AgenticTesting: + """Exercises the agent runtime: compiles a randomized storyteller agent, serves its + stop_when worker from this process, and periodically starts runs that add one + sentence per turn until a critic agent approves the story or max_turns runs out. + Either way the run completes.""" + + def __init__(self, configuration: Configuration) -> None: + self._configuration = configuration + self._model = env_str_or_default("HARNESS_AGENTIC_MODEL", "anthropic/claude-haiku-4-5") + self._max_turns = env_int_or_default("HARNESS_AGENTIC_MAX_TURNS", 20) + self._interval_seconds = env_int_or_default("HARNESS_AGENTIC_INTERVAL_SECONDS", 300) + self._runs_per_concept = env_int_or_default("HARNESS_AGENTIC_RUNS_PER_CONCEPT", 3) + # A fixed seed reproduces the same concept and setting sequence, which is what + # recording once and replaying through mock/mockLLM requires. + self._rng = random.Random(os.environ.get("HARNESS_AGENTIC_SEED") or None) + self._runtime: AgentRuntime | None = None + self._governor: AgenticGovernor | None = None + self._stopped = threading.Event() + + def prepare(self) -> None: + """Deploy the first concept and fork the storyteller's worker processes. Starts no threads.""" + self._runtime = AgentRuntime(configuration=self._configuration) + concept = invent_concept(self._rng) + self._runtime.serve( + build_storyteller(concept, self._model, self._max_turns), blocking=False, + ) + print(f"AgenticTesting: serving {STORYTELLER_NAME} on {self._model}, " + f"first concept: {concept.summary()}") + self._governor = AgenticGovernor( + self._runtime, self._rng, concept, self._model, self._max_turns, + self._interval_seconds, self._runs_per_concept, + ) + + def start(self) -> None: + """Start the governor thread. Call only after every module's prepare().""" + self._governor.start() + + def stop(self) -> None: + if self._governor is not None: + self._governor.stop() + if self._runtime is not None: + self._runtime.shutdown() + self._stopped.set() + + def wait(self) -> None: + """Block until stop() is called.""" + self._stopped.wait() diff --git a/harness/agentic/story_agents.py b/harness/agentic/story_agents.py new file mode 100644 index 000000000..11456b460 --- /dev/null +++ b/harness/agentic/story_agents.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import json +import os +from typing import Any, Optional + +from conductor.ai.agents import Agent, agent_tool + +from agentic.story_concepts import StoryConcept + +# Names stay fixed across concepts: worker task names derive from the agent name +# (e.g. python_storyteller_stop_when), so a stable name means the worker processes +# forked at startup keep serving every later concept and no fork happens mid-run. +# The python_ prefix lets other SDK harnesses share a cluster without collisions. +STORYTELLER_NAME = "python_storyteller" +CRITIC_NAME = "python_story_critic" +CRITIC_TOOL_NAME = "story_critic" + +_VERDICT_SHAPE = '{"done": true or false, "missing": ["short instruction", ...], "note": "one sentence"}' +_MAX_SEARCH_DEPTH = 6 + + +def build_storyteller(concept: StoryConcept, model: str, max_turns: int) -> Agent: + critic = Agent( + name=CRITIC_NAME, + model=model, + instructions=_critic_instructions(concept), + max_tokens=512, + ) + return Agent( + name=STORYTELLER_NAME, + model=model, + instructions=_storyteller_instructions(concept), + tools=[ + agent_tool( + critic, + name=CRITIC_TOOL_NAME, + description=( + "Judge whether the story meets every requirement. Pass the COMPLETE " + "story so far: every sentence, in order." + ), + ) + ], + stop_when=story_is_complete, + max_turns=max_turns, + max_tokens=2048, + ) + + +def story_prompt(setting: str) -> str: + """The user prompt for one run. Shared so every caller sends identical bytes, + which replaying recordings through mock/mockLLM depends on.""" + return f"Write the story. It is set in {setting}." + + +def _storyteller_instructions(concept: StoryConcept) -> str: + return ( + "You are writing a short story one sentence at a time, starting from a blank page.\n\n" + "The finished story must satisfy every one of these requirements:\n" + f"{concept.rubric()}\n\n" + "Each turn:\n" + "1. Write exactly ONE new sentence that continues the story.\n" + f"2. Call {CRITIC_TOOL_NAME} with the COMPLETE story so far: every sentence, in order, " + "including the one you just wrote.\n" + "3. Read the critic's verdict and use its list of missing elements to choose your " + "next sentence.\n\n" + "Never rewrite or remove an earlier sentence. Keep each sentence under 30 words.\n" + f"When the critic says the story is done, do not call {CRITIC_TOOL_NAME} again: " + "reply with the complete final story as plain text." + ) + + +def _critic_instructions(concept: StoryConcept) -> str: + return ( + "You are a strict but fair story editor. You will receive a story draft.\n\n" + "Decide whether the draft satisfies EVERY requirement below and reads as a complete " + "story with a beginning, a middle and an end:\n" + f"{concept.rubric()}\n\n" + "Respond with ONLY a JSON object and no other text, in exactly this shape:\n" + f"{_VERDICT_SHAPE}\n\n" + 'Set "done" to true only if every requirement is clearly met. Otherwise list each ' + 'unmet requirement in "missing" as a short instruction for the next sentence, for ' + 'example "introduce the conflict" or "resolve it through a small act of courage".' + ) + + +def story_is_complete(context: dict) -> bool: + """Loop exit gate: True once the critic's latest verdict says the story is done. + + Runs as a worker in a spawned process, so it must stay a module-level function: + spawn pickling cannot carry closures. ``context["messages"]`` is the LLM's *input* + history, so the newest verdict it can see is the previous turn's -- the loop ends + at most one turn after the critic approves. + """ + iteration = context.get("iteration", 0) + verdict = find_verdict(context.get("messages")) + if verdict is None: + if _is_debug_enabled(): + print(f"[{STORYTELLER_NAME}] turn {iteration}: no verdict yet; " + f"messages={_truncate(context.get('messages'))}") + return False + done = verdict.get("done") is True + print(f"[{STORYTELLER_NAME}] turn {iteration}: latest verdict done={done} " + f"missing={verdict.get('missing')}") + return done + + +def find_verdict(node: Any, depth: int = 0) -> Optional[dict]: + """Return the newest dict carrying a boolean ``done``, searching newest-first. + + The exact shape of a tool result inside the message history is not part of any + published contract, so search the structure rather than index into it. + """ + if depth > _MAX_SEARCH_DEPTH: + return None + if isinstance(node, str): + node = _parse_json_object(node) + if isinstance(node, dict): + if isinstance(node.get("done"), bool): + return node + for value in node.values(): + found = find_verdict(value, depth + 1) + if found is not None: + return found + elif isinstance(node, list): + for item in reversed(node): + found = find_verdict(item, depth + 1) + if found is not None: + return found + return None + + +def _parse_json_object(text: str) -> Optional[dict]: + """Parse a JSON object from model output, tolerating code fences and stray prose.""" + start = text.find("{") + end = text.rfind("}") + if start < 0 or end <= start: + return None + try: + parsed = json.loads(text[start:end + 1]) + except ValueError: + return None + return parsed if isinstance(parsed, dict) else None + + +def _is_debug_enabled() -> bool: + return os.environ.get("HARNESS_AGENTIC_DEBUG", "").strip().lower() in ("1", "true", "yes", "on") + + +def _truncate(value: Any, limit: int = 2000) -> str: + text = repr(value) + return text if len(text) <= limit else text[:limit] + "...(truncated)" diff --git a/harness/agentic/story_concepts.py b/harness/agentic/story_concepts.py new file mode 100644 index 000000000..7356d2a19 --- /dev/null +++ b/harness/agentic/story_concepts.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import random +from dataclasses import dataclass + +# Building blocks for a story concept. Each entry must be something an LLM critic +# can judge reliably from the text alone -- avoid counting rules ("exactly five +# words"), which models judge badly and which stop the story loop converging. + +PROTAGONISTS = [ + "a lighthouse keeper who has never left the island", + "a retired chess champion", + "a stray dog in a busy city", + "an AI that runs a small bakery", + "a cartographer mapping a coastline that keeps changing", + "a night-shift nurse", + "a child who collects lost keys", + "a beekeeper whose hives have gone silent", +] + +CONFLICTS = [ + "a storm cuts them off from everyone they rely on", + "an old rival returns with a claim on what they love", + "they discover a secret someone close to them has kept", + "they must choose between two people who both need them", + "something precious to them goes missing", + "a promise they made years ago comes due", +] + +RESOLUTIONS = [ + "through an unlikely friendship", + "by giving up something they value", + "through a small act of courage", + "by accepting what cannot be changed", + "with a reversal the reader does not see coming", +] + +CONSTRAINTS = [ + "contains no dialogue", + "takes place within a single day", + "is told in the second person", + "names only one character", + "mentions the weather in its final sentence", +] + +TONES = ["wry", "melancholy", "hopeful", "eerie", "warm"] + +# Per-run input. A concept is compiled once and run several times; the setting +# is what varies between those runs. +SETTINGS = [ + "a night train crossing a border", + "a flooded library", + "a market at dawn", + "a hospital corridor at 3 a.m.", + "a village the day after a festival", + "an orbiting research station", + "a ferry stuck in fog", + "an empty stadium after the final match", +] + + +@dataclass(frozen=True) +class StoryConcept: + protagonist: str + conflict: str + resolution: str + constraint: str + tone: str + + def rubric(self) -> str: + """The requirements the critic judges a draft against.""" + return ( + f"- The protagonist is {self.protagonist}.\n" + f"- The protagonist faces this conflict: {self.conflict}.\n" + f"- The conflict is resolved {self.resolution}.\n" + f"- The story {self.constraint}.\n" + f"- The tone is {self.tone}." + ) + + def summary(self) -> str: + return f"{self.protagonist} | {self.tone} | {self.constraint}" + + +def invent_concept(rng: random.Random) -> StoryConcept: + return StoryConcept( + protagonist=rng.choice(PROTAGONISTS), + conflict=rng.choice(CONFLICTS), + resolution=rng.choice(RESOLUTIONS), + constraint=rng.choice(CONSTRAINTS), + tone=rng.choice(TONES), + ) + + +def pick_setting(rng: random.Random) -> str: + return rng.choice(SETTINGS) diff --git a/harness/harness_env.py b/harness/harness_env.py new file mode 100644 index 000000000..04cb38405 --- /dev/null +++ b/harness/harness_env.py @@ -0,0 +1,29 @@ +from __future__ import annotations + +import os + +_TRUE_VALUES = ("1", "true", "yes", "on") +_FALSE_VALUES = ("0", "false", "no", "off") + + +def env_int_or_default(key: str, default: int) -> int: + s = os.environ.get(key, "") + if not s: + return default + try: + return int(s) + except ValueError: + return default + + +def env_str_or_default(key: str, default: str) -> str: + return os.environ.get(key, "") or default + + +def env_bool_or_default(key: str, default: bool) -> bool: + s = os.environ.get(key, "").strip().lower() + if s in _TRUE_VALUES: + return True + if s in _FALSE_VALUES: + return False + return default diff --git a/harness/main.py b/harness/main.py index c7e38b767..1ebe17f0c 100644 --- a/harness/main.py +++ b/harness/main.py @@ -4,67 +4,14 @@ import signal import sys -from conductor.client.automator.task_handler import TaskHandler from conductor.client.configuration.configuration import Configuration from conductor.client.configuration.settings.metrics_settings import MetricsSettings -from conductor.client.http.models.task_def import TaskDef from conductor.client.orkes_clients import OrkesClients from conductor.client.telemetry.metrics_factory import create_metrics_collector_for_parent -from conductor.client.workflow.conductor_workflow import ConductorWorkflow -from conductor.client.workflow.task.simple_task import SimpleTask - -from simulated_task_worker import SimulatedTaskWorker -from workflow_governor import WorkflowGovernor -from workflow_status_probe import WorkflowStatusProbe - -WORKFLOW_NAME = "python_simulated_tasks_workflow" - -SIMULATED_WORKERS = [ - ("python_worker_0", "quickpulse", 1), - ("python_worker_1", "whisperlink", 2), - ("python_worker_2", "shadowfetch", 3), - ("python_worker_3", "ironforge", 4), - ("python_worker_4", "deepcrawl", 5), -] - - -def env_int_or_default(key: str, default: int) -> int: - s = os.environ.get(key, "") - if not s: - return default - try: - return int(s) - except ValueError: - return default - - -def register_metadata(clients: OrkesClients) -> None: - metadata_client = clients.get_metadata_client() - workflow_executor = clients.get_workflow_executor() - - for task_name, codename, sleep_seconds in SIMULATED_WORKERS: - task_def = TaskDef( - name=task_name, - description=f"Python SDK harness simulated task ({codename}, default delay {sleep_seconds}s)", - retry_count=1, - timeout_seconds=300, - response_timeout_seconds=300, - ) - metadata_client.register_task_def(task_def) - - wf = ConductorWorkflow( - executor=workflow_executor, - name=WORKFLOW_NAME, - version=1, - description="Python SDK harness simulated task workflow", - ) - wf.owner_email("python-sdk-harness@conductor.io") - - for task_name, codename, _ in SIMULATED_WORKERS: - wf.add(SimpleTask(task_def_name=task_name, task_reference_name=codename)) - wf.register(overwrite=True) - print(f"Registered workflow {WORKFLOW_NAME} with {len(SIMULATED_WORKERS)} tasks") +from agentic import AgenticTesting +from harness_env import env_bool_or_default, env_int_or_default +from sdk_testing import SDKTesting def main() -> None: @@ -86,34 +33,17 @@ def main() -> None: print(f"Prometheus metrics server started on port {metrics_port} ({metrics_collector.collector_name()} metrics)") clients = OrkesClients(configuration, metrics_collector=metrics_collector) - register_metadata(clients) - - workflows_per_sec = env_int_or_default("HARNESS_WORKFLOWS_PER_SEC", 2) - batch_size = env_int_or_default("HARNESS_BATCH_SIZE", 20) - poll_interval_ms = env_int_or_default("HARNESS_POLL_INTERVAL_MS", 100) - - workers = [] - for task_name, codename, sleep_seconds in SIMULATED_WORKERS: - worker = SimulatedTaskWorker(task_name, codename, sleep_seconds, batch_size, poll_interval_ms) - workers.append(worker) + sdk_testing = None + if env_bool_or_default("HARNESS_SDK_TESTING_ENABLED", True): + sdk_testing = SDKTesting(configuration, clients, metrics_settings) - task_handler = TaskHandler( - workers=workers, - configuration=configuration, - scan_for_annotated_workers=False, - metrics_settings=metrics_settings, - ) - - workflow_executor = clients.get_workflow_executor() - workflow_client = clients.get_workflow_client() + agentic_testing = None + if env_bool_or_default("HARNESS_AGENTIC_TESTING_ENABLED", False): + agentic_testing = AgenticTesting(configuration) - probe_rate = env_int_or_default("HARNESS_PROBE_RATE_PER_SEC", 0) - probe = WorkflowStatusProbe(workflow_client, probe_rate) - - governor = WorkflowGovernor( - workflow_executor, WORKFLOW_NAME, workflows_per_sec, - id_sink=probe.offer, - ) + if sdk_testing is None and agentic_testing is None: + print("Nothing to run: HARNESS_SDK_TESTING_ENABLED and HARNESS_AGENTIC_TESTING_ENABLED are both off") + sys.exit(1) main_pid = os.getpid() shutting_down = False @@ -124,23 +54,34 @@ def shutdown(signum, frame): return shutting_down = True print("Shutting down...") - probe.shutdown() - governor.stop() - task_handler.stop_processes() + if agentic_testing is not None: + agentic_testing.stop() + if sdk_testing is not None: + sdk_testing.stop() sys.exit(0) signal.signal(signal.SIGINT, shutdown) signal.signal(signal.SIGTERM, shutdown) - # Fork worker processes BEFORE starting the governor thread. The parent's - # MetricsCollector uses prometheus_client's multiprocess mode, which guards - # its mmapped value files with a threading.Lock. If the governor thread is - # already running at fork() time, a child can inherit that lock in a held - # state and deadlock on its first metric write. - task_handler.start_processes() - probe.start() - governor.start() - task_handler.join_processes() + # Fork every module's worker processes BEFORE starting any module's threads. + # The parent's MetricsCollector uses prometheus_client's multiprocess mode, + # which guards its mmapped value files with a threading.Lock. If a governor + # thread is already running at fork() time, a child can inherit that lock in + # a held state and deadlock on its first metric write. + if sdk_testing is not None: + sdk_testing.prepare() + if agentic_testing is not None: + agentic_testing.prepare() + + if sdk_testing is not None: + sdk_testing.start() + if agentic_testing is not None: + agentic_testing.start() + + if sdk_testing is not None: + sdk_testing.wait() + else: + agentic_testing.wait() if __name__ == "__main__": diff --git a/harness/sdk_testing.py b/harness/sdk_testing.py new file mode 100644 index 000000000..b860aa695 --- /dev/null +++ b/harness/sdk_testing.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +from conductor.client.automator.task_handler import TaskHandler +from conductor.client.configuration.configuration import Configuration +from conductor.client.configuration.settings.metrics_settings import MetricsSettings +from conductor.client.http.models.task_def import TaskDef +from conductor.client.orkes_clients import OrkesClients +from conductor.client.workflow.conductor_workflow import ConductorWorkflow +from conductor.client.workflow.task.simple_task import SimpleTask + +from harness_env import env_int_or_default +from simulated_task_worker import SimulatedTaskWorker +from workflow_governor import WorkflowGovernor +from workflow_status_probe import WorkflowStatusProbe + +WORKFLOW_NAME = "python_simulated_tasks_workflow" + +SIMULATED_WORKERS = [ + ("python_worker_0", "quickpulse", 1), + ("python_worker_1", "whisperlink", 2), + ("python_worker_2", "shadowfetch", 3), + ("python_worker_3", "ironforge", 4), + ("python_worker_4", "deepcrawl", 5), +] + + +def register_metadata(clients: OrkesClients) -> None: + metadata_client = clients.get_metadata_client() + workflow_executor = clients.get_workflow_executor() + + for task_name, codename, sleep_seconds in SIMULATED_WORKERS: + task_def = TaskDef( + name=task_name, + description=f"Python SDK harness simulated task ({codename}, default delay {sleep_seconds}s)", + retry_count=1, + timeout_seconds=300, + response_timeout_seconds=300, + ) + metadata_client.register_task_def(task_def) + + wf = ConductorWorkflow( + executor=workflow_executor, + name=WORKFLOW_NAME, + version=1, + description="Python SDK harness simulated task workflow", + ) + wf.owner_email("python-sdk-harness@conductor.io") + + for task_name, codename, _ in SIMULATED_WORKERS: + wf.add(SimpleTask(task_def_name=task_name, task_reference_name=codename)) + + wf.register(overwrite=True) + print(f"Registered workflow {WORKFLOW_NAME} with {len(SIMULATED_WORKERS)} tasks") + + +class SDKTesting: + """Exercises the core SDK: registers five simulated tasks and a workflow that + chains them, serves those tasks from this process, and starts workflows at a + fixed rate so the SDK is continuously both workflow starter and task worker.""" + + def __init__( + self, + configuration: Configuration, + clients: OrkesClients, + metrics_settings: MetricsSettings, + ) -> None: + self._configuration = configuration + self._clients = clients + self._metrics_settings = metrics_settings + self._task_handler: TaskHandler | None = None + self._probe: WorkflowStatusProbe | None = None + self._governor: WorkflowGovernor | None = None + + def prepare(self) -> None: + """Register metadata and fork the worker processes. Starts no threads.""" + register_metadata(self._clients) + + workflows_per_sec = env_int_or_default("HARNESS_WORKFLOWS_PER_SEC", 2) + batch_size = env_int_or_default("HARNESS_BATCH_SIZE", 20) + poll_interval_ms = env_int_or_default("HARNESS_POLL_INTERVAL_MS", 100) + + workers = [] + for task_name, codename, sleep_seconds in SIMULATED_WORKERS: + worker = SimulatedTaskWorker(task_name, codename, sleep_seconds, batch_size, poll_interval_ms) + workers.append(worker) + + self._task_handler = TaskHandler( + workers=workers, + configuration=self._configuration, + scan_for_annotated_workers=False, + metrics_settings=self._metrics_settings, + ) + + workflow_executor = self._clients.get_workflow_executor() + workflow_client = self._clients.get_workflow_client() + + probe_rate = env_int_or_default("HARNESS_PROBE_RATE_PER_SEC", 0) + self._probe = WorkflowStatusProbe(workflow_client, probe_rate) + + self._governor = WorkflowGovernor( + workflow_executor, WORKFLOW_NAME, workflows_per_sec, + id_sink=self._probe.offer, + ) + + self._task_handler.start_processes() + + def start(self) -> None: + """Start the probe and governor threads. Call only after every module's prepare().""" + self._probe.start() + self._governor.start() + + def stop(self) -> None: + if self._probe is not None: + self._probe.shutdown() + if self._governor is not None: + self._governor.stop() + if self._task_handler is not None: + self._task_handler.stop_processes() + + def wait(self) -> None: + """Block until the worker processes exit.""" + self._task_handler.join_processes() diff --git a/harness/try_agentic.py b/harness/try_agentic.py new file mode 100644 index 000000000..569126876 --- /dev/null +++ b/harness/try_agentic.py @@ -0,0 +1,265 @@ +"""Run the agentic storyteller once, locally, without the harness governors. + + python3 harness/try_agentic.py plan # server compiles the agent; nothing runs + python3 harness/try_agentic.py run # one story, live progress, then a report + +Server and auth come from CONDUCTOR_SERVER_URL, CONDUCTOR_AUTH_KEY and +CONDUCTOR_AUTH_SECRET. Everything else is a flag; see --help on each command. + +Runs register the agent under the same name the harness uses (python_storyteller), +so don't point this at a cluster where the harness's agentic testing is running. +""" +from __future__ import annotations + +import argparse +import json +import os +import random +import secrets +import sys +import textwrap +import threading +from typing import Any, Optional + +from conductor.ai.agents import AgentRuntime +from conductor.client.configuration.configuration import Configuration + +from agentic.story_agents import CRITIC_TOOL_NAME, build_storyteller, find_verdict, story_prompt +from agentic.story_concepts import StoryConcept, invent_concept, pick_setting +from harness_env import env_str_or_default + +DEFAULT_MODEL = "anthropic/claude-haiku-4-5" +WRAP = 88 + + +def main() -> int: + args = parse_args() + if getattr(args, "debug", False): + # Read by the stop_when worker, which runs in a separate process. + os.environ["HARNESS_AGENTIC_DEBUG"] = "true" + + seed = args.seed or secrets.token_hex(4) + rng = random.Random(seed) + concept = invent_concept(rng) + configuration = Configuration() + + print(f"server: {configuration.host}") + print(f"model: {args.model}") + print(f"seed: {seed} (repeat this concept with --seed {seed})") + print(f"concept:\n{textwrap.indent(concept.rubric(), ' ')}") + + runtime = AgentRuntime(configuration=configuration) + try: + return args.handler(args, runtime, rng, concept) + finally: + runtime.shutdown() + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + commands = parser.add_subparsers(dest="command", required=True) + + common = argparse.ArgumentParser(add_help=False) + common.add_argument("--model", default=env_str_or_default("HARNESS_AGENTIC_MODEL", DEFAULT_MODEL), + help="provider/model; on Orkes the provider is the AI integration's name " + f"(default: $HARNESS_AGENTIC_MODEL or {DEFAULT_MODEL})") + common.add_argument("--seed", help="seed for the concept and setting (default: random, printed)") + common.add_argument("--max-turns", type=int, default=20, help="turn limit (default: 20)") + + plan = commands.add_parser("plan", parents=[common], + help="compile the storyteller on the server and show the result") + plan.add_argument("--out", help="also write the full compile response to this JSON file") + plan.set_defaults(handler=run_plan) + + run = commands.add_parser("run", parents=[common], help="run one story and report the result") + run.add_argument("--setting", help="story setting (default: random pick)") + run.add_argument("--timeout", type=int, default=600, + help="seconds before the run is cancelled (default: 600)") + run.add_argument("--quiet", action="store_true", help="no live progress, just the report") + run.add_argument("--debug", action="store_true", + help="the stop_when worker prints raw message history when it finds no verdict") + run.set_defaults(handler=run_story) + + return parser.parse_args() + + +# ── plan ──────────────────────────────────────────────────────────────── + + +def run_plan(args: argparse.Namespace, runtime: AgentRuntime, rng: random.Random, + concept: StoryConcept) -> int: + agent = build_storyteller(concept, args.model, args.max_turns) + try: + response = runtime.plan(agent) + except Exception as e: + print(f"\nThe server did not compile the agent: {type(e).__name__}: {e}") + if "404" in str(e): + print("A 404 usually means the agent API is off (conductor.integrations.ai.enabled).") + elif "Connection" in str(e): + print("Nothing answered at that address. Check CONDUCTOR_SERVER_URL.") + return 1 + + workflow = _field(response, "workflowDef", "workflow_def") or {} + workers = _field(response, "requiredWorkers", "required_workers") or [] + print(f"\nagent API: OK, compiled {_field(workflow, 'name')} v{_field(workflow, 'version')}") + print(f"workers this process must serve: {', '.join(workers) or '(none)'}\n") + _print_tasks(_field(workflow, "tasks") or [], depth=0) + + if args.out: + with open(args.out, "w") as f: + json.dump(response, f, indent=2, default=str) + print(f"\nfull compile response written to {args.out}") + return 0 + + +def _print_tasks(tasks: list, depth: int) -> None: + pad = " " * depth + for task in tasks: + task_type = _field(task, "type") or "?" + line = f"{pad}- {task_type:<22} {_field(task, 'taskReferenceName') or ''}" + if task_type == "LLM_CHAT_COMPLETE": + line += _describe_llm_task(_field(task, "inputParameters") or {}) + print(line) + for case, branch in (_field(task, "decisionCases") or {}).items(): + print(f"{pad} case {case}:") + _print_tasks(branch, depth + 3) + if _field(task, "defaultCase"): + print(f"{pad} default:") + _print_tasks(_field(task, "defaultCase"), depth + 3) + _print_tasks(_field(task, "loopOver") or [], depth + 1) + for branch in _field(task, "forkTasks") or []: + _print_tasks(branch, depth + 1) + sub = (_field(task, "subWorkflowParam") or {}).get("workflowDefinition") + if isinstance(sub, dict): + print(f"{pad} sub-workflow {sub.get('name')}:") + _print_tasks(sub.get("tasks") or [], depth + 3) + + +def _describe_llm_task(inputs: dict) -> str: + text = (f" [{inputs.get('llmProvider')}/{inputs.get('model')}" + f" jsonOutput={bool(inputs.get('jsonOutput'))}" + f" outputSchema={'yes' if inputs.get('outputSchema') else 'no'}") + tools = inputs.get("tools") + if isinstance(tools, list) and tools: + text += " tools=" + ", ".join(f"{t.get('name')}({t.get('type')})" for t in tools) + return text + "]" + + +# ── run ───────────────────────────────────────────────────────────────── + + +def run_story(args: argparse.Namespace, runtime: AgentRuntime, rng: random.Random, + concept: StoryConcept) -> int: + setting = args.setting or pick_setting(rng) + print(f"setting: {setting}\n") + + agent = build_storyteller(concept, args.model, args.max_turns) + handle = runtime.start(agent, story_prompt(setting)) + print(f"execution: {handle.execution_id}\n") + + # handle.stream() has no timeout of its own (and run(on_event=...) ignores the one it + # accepts), so read it on a thread and enforce the deadline here. + stream = handle.stream() + progress = _Progress(quiet=args.quiet) + finished = threading.Event() + + def consume() -> None: + try: + for event in stream: + progress.on_event(event) + except Exception as e: + print(f" event stream failed: {type(e).__name__}: {e}") + finally: + finished.set() + + threading.Thread(target=consume, name="try-agentic-stream", daemon=True).start() + try: + if not finished.wait(timeout=args.timeout): + handle.cancel(f"try_agentic.py: no result after {args.timeout}s") + print(f"\nNo result after {args.timeout}s; cancelled {handle.execution_id}.") + return 1 + except KeyboardInterrupt: + handle.cancel("try_agentic.py: interrupted") + print(f"\nInterrupted; cancelled {handle.execution_id}.") + return 130 + + return _report(handle, stream, progress) + + +class _Progress: + """Turns stream events into one line per sentence and one per critic verdict.""" + + def __init__(self, quiet: bool) -> None: + self.quiet = quiet + self.story = "" + self.verdicts: list[Optional[dict]] = [] + + def on_event(self, event: Any) -> None: + kind = str(getattr(event.type, "value", event.type)) + is_critic = CRITIC_TOOL_NAME in (event.tool_name or "") + if kind == "tool_call" and is_critic: + story = str((event.args or {}).get("request", "")) + added = story[len(self.story):].strip() if story.startswith(self.story) else story + self.story = story + self._say(f" turn {len(self.verdicts) + 1:>2} + {added}") + elif kind == "tool_result" and is_critic: + verdict = find_verdict(event.result) + self.verdicts.append(verdict) + self._say(f" critic: {_describe_verdict(verdict)}") + elif kind == "error": + self._say(f" error: {event.content}") + + def _say(self, line: str) -> None: + if not self.quiet: + limit = WRAP + 40 + print(line if len(line) <= limit else line[:limit - 4] + " ...") + + +def _report(handle: Any, stream: Any, progress: _Progress) -> int: + try: + result = stream.get_result() + status = str(getattr(result.status, "value", result.status)) + output, usage, error = result.output, result.token_usage, result.error + except Exception: + snapshot = handle.get_status() + status, output, usage, error = snapshot.status, snapshot.output, None, snapshot.reason + + approved = any(v is not None and v.get("done") is True for v in progress.verdicts) + print("\n" + "=" * WRAP) + print(f"status: {status}") + print(f"execution: {handle.execution_id}") + print(f"critic calls: {len(progress.verdicts)} approved: " + f"{'yes' if approved else 'no (stopped at the turn limit, or failed)'}") + if usage is not None: + print(f"tokens: {usage.prompt_tokens} in, {usage.completion_tokens} out, " + f"{usage.total_tokens} total") + if error: + print(f"error: {error}") + + story = output if isinstance(output, str) and output.strip() else progress.story + print("\nstory:\n") + print(textwrap.fill(story, WRAP) if story else "(no story text came back)") + return 0 if status.upper() == "COMPLETED" else 1 + + +def _describe_verdict(verdict: Optional[dict]) -> str: + if verdict is None: + return "no JSON verdict found in its reply" + if verdict.get("done") is True: + return "done" + return f"not done, missing: {verdict.get('missing')}" + + +def _field(obj: Any, *names: str) -> Any: + """Read a key from a dict response, or an attribute from a model object.""" + for name in names: + value = obj.get(name) if isinstance(obj, dict) else getattr(obj, name, None) + if value is not None: + return value + return None + + +if __name__ == "__main__": + # The guard is required: on macOS worker processes are spawned, and each one + # re-imports this file as a module. + sys.exit(main()) From c7d5ae9c1d1eed3d28dbc80f1227778b2b9692fa Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Thu, 29 Oct 2026 11:55:15 -0600 Subject: [PATCH 2/8] feat(harness): execution links, --open and --verbose for try_agentic.py - Print a link to the execution in the UI when a run starts and in the final report. The UI address is the server URL without /api, or --ui-url / HARNESS_UI_URL when they differ. - --open opens that link in the browser. - --verbose shows the SDK's own INFO logs (not the HTTP library's, which can print request headers carrying the auth token). Added to see which workers the server asks the SDK to run: on Orkes 5.6.0-rc4 a run started from the agent object never got a stop_when worker and waited forever. - story_prompt() no longer doubles the period after settings that end in one ("at 3 a.m.."). Still untested against a real server end to end. Co-Authored-By: Claude Opus 5.5 --- harness/README.md | 2 ++ harness/agentic/story_agents.py | 3 ++- harness/try_agentic.py | 38 ++++++++++++++++++++++++++++++--- 3 files changed, 39 insertions(+), 4 deletions(-) diff --git a/harness/README.md b/harness/README.md index e5cbc0a63..d6fdba050 100644 --- a/harness/README.md +++ b/harness/README.md @@ -153,6 +153,8 @@ python3 harness/try_agentic.py run python3 harness/try_agentic.py run --seed 3f9a1c2e --setting "a flooded library" --max-turns 10 ``` +`run` prints a link to the execution in the UI as soon as it starts (Cmd-click it in VS Code or iTerm2), and `--open` opens it in your browser. The link assumes the UI is served at the server URL without `/api`; if it isn't, pass `--ui-url` or set `HARNESS_UI_URL`. + A run is cancelled on Ctrl-C or after `--timeout` seconds (default 600), so it doesn't keep calling the LLM. It registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. On Orkes, `--model` takes the AI integration's name before the slash. ### Agentic Environment Variables diff --git a/harness/agentic/story_agents.py b/harness/agentic/story_agents.py index 11456b460..70217ca0b 100644 --- a/harness/agentic/story_agents.py +++ b/harness/agentic/story_agents.py @@ -50,7 +50,8 @@ def build_storyteller(concept: StoryConcept, model: str, max_turns: int) -> Agen def story_prompt(setting: str) -> str: """The user prompt for one run. Shared so every caller sends identical bytes, which replaying recordings through mock/mockLLM depends on.""" - return f"Write the story. It is set in {setting}." + # Settings like "... at 3 a.m." already end in a period. + return f"Write the story. It is set in {setting.rstrip('.')}." def _storyteller_instructions(concept: StoryConcept) -> str: diff --git a/harness/try_agentic.py b/harness/try_agentic.py index 569126876..8e5ceef49 100644 --- a/harness/try_agentic.py +++ b/harness/try_agentic.py @@ -13,12 +13,14 @@ import argparse import json +import logging import os import random import secrets import sys import textwrap import threading +import webbrowser from typing import Any, Optional from conductor.ai.agents import AgentRuntime @@ -34,6 +36,11 @@ def main() -> int: args = parse_args() + if args.verbose: + # Only the SDK's loggers: HTTP-library logs are noisy at INFO, and at DEBUG + # they can print request headers, which carry the auth token. + logging.basicConfig(level=logging.WARNING, format="%(asctime)s %(levelname)s %(name)s: %(message)s") + logging.getLogger("conductor").setLevel(logging.INFO) if getattr(args, "debug", False): # Read by the stop_when worker, which runs in a separate process. os.environ["HARNESS_AGENTIC_DEBUG"] = "true" @@ -42,6 +49,7 @@ def main() -> int: rng = random.Random(seed) concept = invent_concept(rng) configuration = Configuration() + args.server = configuration.host print(f"server: {configuration.host}") print(f"model: {args.model}") @@ -65,6 +73,8 @@ def parse_args() -> argparse.Namespace: f"(default: $HARNESS_AGENTIC_MODEL or {DEFAULT_MODEL})") common.add_argument("--seed", help="seed for the concept and setting (default: random, printed)") common.add_argument("--max-turns", type=int, default=20, help="turn limit (default: 20)") + common.add_argument("--verbose", action="store_true", + help="show the SDK's INFO logs, e.g. which workers the server asked for") plan = commands.add_parser("plan", parents=[common], help="compile the storyteller on the server and show the result") @@ -76,6 +86,11 @@ def parse_args() -> argparse.Namespace: run.add_argument("--timeout", type=int, default=600, help="seconds before the run is cancelled (default: 600)") run.add_argument("--quiet", action="store_true", help="no live progress, just the report") + run.add_argument("--open", action="store_true", + help="open the execution in your browser as soon as the run starts") + run.add_argument("--ui-url", default=os.environ.get("HARNESS_UI_URL"), + help="UI address for execution links (default: $HARNESS_UI_URL, else the " + "server URL without /api)") run.add_argument("--debug", action="store_true", help="the stop_when worker prints raw message history when it finds no verdict") run.set_defaults(handler=run_story) @@ -155,7 +170,12 @@ def run_story(args: argparse.Namespace, runtime: AgentRuntime, rng: random.Rando agent = build_storyteller(concept, args.model, args.max_turns) handle = runtime.start(agent, story_prompt(setting)) - print(f"execution: {handle.execution_id}\n") + link = _execution_url(args.ui_url or args.server, handle.execution_id) + print(f"execution: {handle.execution_id}") + print(f"view: {link}") + if args.open and not webbrowser.open(link): + print(" (could not open a browser; Cmd-click the link instead)") + print() # handle.stream() has no timeout of its own (and run(on_event=...) ignores the one it # accepts), so read it on a thread and enforce the deadline here. @@ -183,7 +203,7 @@ def consume() -> None: print(f"\nInterrupted; cancelled {handle.execution_id}.") return 130 - return _report(handle, stream, progress) + return _report(handle, stream, progress, link) class _Progress: @@ -215,7 +235,7 @@ def _say(self, line: str) -> None: print(line if len(line) <= limit else line[:limit - 4] + " ...") -def _report(handle: Any, stream: Any, progress: _Progress) -> int: +def _report(handle: Any, stream: Any, progress: _Progress, link: str) -> int: try: result = stream.get_result() status = str(getattr(result.status, "value", result.status)) @@ -228,6 +248,7 @@ def _report(handle: Any, stream: Any, progress: _Progress) -> int: print("\n" + "=" * WRAP) print(f"status: {status}") print(f"execution: {handle.execution_id}") + print(f"view: {link}") print(f"critic calls: {len(progress.verdicts)} approved: " f"{'yes' if approved else 'no (stopped at the turn limit, or failed)'}") if usage is not None: @@ -250,6 +271,17 @@ def _describe_verdict(verdict: Optional[dict]) -> str: return f"not done, missing: {verdict.get('missing')}" +def _execution_url(base: str, execution_id: str) -> str: + """UI link for an execution. Agent runs use the normal execution page in both the + OSS and Orkes UIs; the agent view is a tab on it. Printed as a plain URL, which + VS Code's terminal and iTerm2 make Cmd-clickable and which survives logs and + copy-paste.""" + base = base.rstrip("/") + if base.endswith("/api"): + base = base[: -len("/api")] + return f"{base}/execution/{execution_id}" + + def _field(obj: Any, *names: str) -> Any: """Read a key from a dict response, or an attribute from a model object.""" for name in names: From 2b8daee68988dca83707d5ac5028635f94ee6cfb Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Thu, 29 Oct 2026 13:13:01 -0600 Subject: [PATCH 3/8] start agent in a way that spawns stop_when worker --- harness/README.md | 6 +++++- harness/try_agentic.py | 15 +++++++++++++-- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/harness/README.md b/harness/README.md index d6fdba050..6c19bd202 100644 --- a/harness/README.md +++ b/harness/README.md @@ -155,7 +155,11 @@ python3 harness/try_agentic.py run --seed 3f9a1c2e --setting "a flooded library" `run` prints a link to the execution in the UI as soon as it starts (Cmd-click it in VS Code or iTerm2), and `--open` opens it in your browser. The link assumes the UI is served at the server URL without `/api`; if it isn't, pass `--ui-url` or set `HARNESS_UI_URL`. -A run is cancelled on Ctrl-C or after `--timeout` seconds (default 600), so it doesn't keep calling the LLM. It registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. On Orkes, `--model` takes the AI integration's name before the slash. +A run is cancelled on Ctrl-C or after `--timeout` seconds (default 600), so it doesn't keep calling the LLM. + +`run` starts things the way the harness does: `serve()` saves the agent and starts every worker it declares, then the run starts the saved agent by name. + +Passing the agent object straight to `start()` doesn't work for this agent. The SDK then starts worker processes only if `_has_worker_tools()` is true, and that check ignores `stop_when`. The storyteller's only tool is the critic, which runs on the server, so its `stop_when` worker is registered but never started, and the run waits forever. It registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. On Orkes, `--model` takes the AI integration's name before the slash. ### Agentic Environment Variables diff --git a/harness/try_agentic.py b/harness/try_agentic.py index 8e5ceef49..1ada5c71a 100644 --- a/harness/try_agentic.py +++ b/harness/try_agentic.py @@ -26,7 +26,13 @@ from conductor.ai.agents import AgentRuntime from conductor.client.configuration.configuration import Configuration -from agentic.story_agents import CRITIC_TOOL_NAME, build_storyteller, find_verdict, story_prompt +from agentic.story_agents import ( + CRITIC_TOOL_NAME, + STORYTELLER_NAME, + build_storyteller, + find_verdict, + story_prompt, +) from agentic.story_concepts import StoryConcept, invent_concept, pick_setting from harness_env import env_str_or_default @@ -169,7 +175,12 @@ def run_story(args: argparse.Namespace, runtime: AgentRuntime, rng: random.Rando print(f"setting: {setting}\n") agent = build_storyteller(concept, args.model, args.max_turns) - handle = runtime.start(agent, story_prompt(setting)) + # The harness's way: serve() saves the agent and starts every worker the agent declares, + # then the run starts the saved agent by name. Don't pass the agent object to start(): + # the SDK then starts worker processes only if _has_worker_tools() is true, which + # ignores stop_when, so this agent's stop_when worker would never run. + runtime.serve(agent, blocking=False) + handle = runtime.start(STORYTELLER_NAME, story_prompt(setting), version=1) link = _execution_url(args.ui_url or args.server, handle.execution_id) print(f"execution: {handle.execution_id}") print(f"view: {link}") From 399b2f1d36db084aed7e7594a6918c9e7fc3ec6a Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Fri, 30 Oct 2026 10:17:21 -0600 Subject: [PATCH 4/8] feat(harness): --stop-same-turn and a report built from the execution stop_when decides whether the storyteller's loop ends, and that decision depends on the result of a tool call: the critic's verdict. As compiled, stop_when reads the LLM call's input messages, a snapshot taken before that turn's critic call, so it sees each verdict one turn late and the loop runs one extra, most expensive storyteller call after approval. - agentic/stop_wiring.py: after deploy, re-point stop_when's messages input at ${workflow.variables._last_tool_results}, which already holds every tool result so far and is written after the critic and before stop_when. Edits the raw definition JSON so no field is lost, refuses to write a definition without its agent markers in metadata, and re-reads it to confirm. Must be re-applied after every deploy; the proper fix is in the server's compiler. - try_agentic.py run --stop-same-turn applies it before starting. - agentic/run_report.py: build the report from the execution record, not streamed events (Orkes 5.6.0-rc4 streams no tool-call events): each turn's sentence and verdict, tokens for storyteller and critic, which stop_when wiring the run used, and the story the critic approved rather than the storyteller's final reply, which can add commentary. Reads go through call_api(response_type="object") because the SDK's raw mode (_preload_content=False) crashes on its httpx transport. Checked against a fake server through the real SDK and against saved executions from real runs; not yet run on a cluster. Co-Authored-By: Claude Opus 5.5 --- harness/README.md | 17 ++++- harness/agentic/run_report.py | 122 +++++++++++++++++++++++++++++++++ harness/agentic/stop_wiring.py | 102 +++++++++++++++++++++++++++ harness/try_agentic.py | 108 +++++++++++++++++++++-------- 4 files changed, 319 insertions(+), 30 deletions(-) create mode 100644 harness/agentic/run_report.py create mode 100644 harness/agentic/stop_wiring.py diff --git a/harness/README.md b/harness/README.md index 6c19bd202..3e4eaf65f 100644 --- a/harness/README.md +++ b/harness/README.md @@ -146,20 +146,33 @@ Open the execution in the Conductor UI to see each storyteller turn and critic c # Confirms the agent API is on and shows the compiled workflow. python3 harness/try_agentic.py plan -# Run one story: one line per sentence and critic verdict, then a report. +# Run one story, then report each turn's sentence and critic verdict, tokens, and the story. python3 harness/try_agentic.py run # Repeat a concept and setting, with a shorter limit python3 harness/try_agentic.py run --seed 3f9a1c2e --setting "a flooded library" --max-turns 10 + +# Stop on the turn the critic approves, saving one storyteller call (see below) +python3 harness/try_agentic.py run --stop-same-turn ``` +The report is built from the execution record after the run, not from streamed events: Orkes 5.6.0-rc4 streams no tool-call events, so live per-turn lines appear only on servers that do. The story it prints is the text the critic approved, exactly as the critic judged it. The storyteller's own final reply can add commentary such as a checklist of the requirements. + `run` prints a link to the execution in the UI as soon as it starts (Cmd-click it in VS Code or iTerm2), and `--open` opens it in your browser. The link assumes the UI is served at the server URL without `/api`; if it isn't, pass `--ui-url` or set `HARNESS_UI_URL`. A run is cancelled on Ctrl-C or after `--timeout` seconds (default 600), so it doesn't keep calling the LLM. `run` starts things the way the harness does: `serve()` saves the agent and starts every worker it declares, then the run starts the saved agent by name. -Passing the agent object straight to `start()` doesn't work for this agent. The SDK then starts worker processes only if `_has_worker_tools()` is true, and that check ignores `stop_when`. The storyteller's only tool is the critic, which runs on the server, so its `stop_when` worker is registered but never started, and the run waits forever. It registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. On Orkes, `--model` takes the AI integration's name before the slash. +Passing the agent object straight to `start()` doesn't work for this agent. The SDK then starts worker processes only if `_has_worker_tools()` is true, and that check ignores `stop_when`. The storyteller's only tool is the critic, which runs on the server, so its `stop_when` worker is registered but never started, and the run waits forever. + +`run` registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. On Orkes, `--model` takes the AI integration's name before the slash. + +#### `--stop-same-turn` + +`stop_when` decides whether the loop ends, and that decision depends on the critic's verdict: the result of a tool call. As compiled, `stop_when` reads the storyteller LLM call's input messages, a snapshot taken before that turn's critic call, so it sees each verdict one turn late and the loop runs one extra storyteller call after the critic approves. That call carries the whole conversation, so it is the most expensive turn. + +`--stop-same-turn` edits the stored workflow definition after `serve()` deploys it and before the run starts: it points `stop_when` at `_last_tool_results`, a workflow variable that already holds every tool result so far and is updated after the critic and before `stop_when`. The loop then stops on the approving turn. The edit is made on the raw JSON so no field is lost, it refuses to write a definition without its agent markers in `metadata`, and it re-reads the definition to confirm. Every deploy regenerates the definition, so the edit has to be re-applied each time; the report's `stop_when:` line says which wiring the run actually used. The proper fix is in the server's compiler: pass this turn's tool results to `stop_when`. See `agentic/stop_wiring.py`. ### Agentic Environment Variables diff --git a/harness/agentic/run_report.py b/harness/agentic/run_report.py new file mode 100644 index 000000000..9f7b37bd2 --- /dev/null +++ b/harness/agentic/run_report.py @@ -0,0 +1,122 @@ +"""Summarize a finished storyteller run from its execution record. + +Reads the execution itself rather than streamed events: some servers (Orkes 5.6.0-rc4) don't +stream tool-call events, so a report built from events misses every critic call. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any, List, Optional, Tuple + +from agentic.stop_wiring import find_stop_when, stop_when_reads_same_turn +from agentic.story_agents import CRITIC_NAME, CRITIC_TOOL_NAME, STORYTELLER_NAME, find_verdict + +# The storyteller sometimes labels the story it sends the critic ("COMPLETE STORY SO FAR:"). +_LABEL = re.compile(r"^\s*(?:the\s+)?(?:complete\s+)?story(?:\s+so\s+far)?\s*:\s*", re.IGNORECASE) + + +@dataclass +class Turn: + number: int + finish_reason: str + story: Optional[str] # the full story sent to the critic; None if the critic wasn't called + added: str # what this turn added to the story + verdict: Optional[dict] + + +@dataclass +class RunSummary: + status: str + turns: List[Turn] + approved_on: Optional[int] + story: str + story_source: str + storyteller_tokens: Tuple[int, int] + critic_execution_ids: List[str] + stop_when_same_turn: Optional[bool] + + +def summarize(execution: dict) -> RunSummary: + tasks = execution.get("tasks") or [] + llm_calls = {t.get("iteration"): t for t in tasks if t.get("taskType") == "LLM_CHAT_COMPLETE"} + critic_calls = {t.get("iteration"): t for t in tasks + if t.get("taskType") == "SUB_WORKFLOW" + and (t.get("inputData") or {}).get("subWorkflowName") == CRITIC_NAME} + + turns: List[Turn] = [] + previous = "" + for number in sorted(n for n in llm_calls if n is not None): + output = llm_calls[number].get("outputData") or {} + story = _story_sent_to_critic(output) + added = "" + if story is not None: + added = story[len(previous):].strip() if story.startswith(previous) else story + previous = story + critic = critic_calls.get(number) + verdict = find_verdict((critic.get("outputData") or {}).get("result")) if critic else None + turns.append(Turn(number, output.get("finishReason") or "", story, added, verdict)) + + approved = next((t for t in turns if t.verdict is not None and t.verdict.get("done") is True), None) + drafts = [t for t in turns if t.story] + if approved is not None and approved.story: + story, source = approved.story, f"approved by the critic on turn {approved.number}" + elif drafts: + story, source = drafts[-1].story, "latest draft; the critic never approved it" + else: + story, source = _workflow_result(execution), "workflow output; the critic was never called" + + return RunSummary( + status=str(execution.get("status") or "UNKNOWN"), + turns=turns, + approved_on=approved.number if approved is not None else None, + story=story, + story_source=source, + storyteller_tokens=llm_tokens(execution), + critic_execution_ids=[critic_calls[n]["subWorkflowId"] for n in sorted(critic_calls) + if critic_calls[n].get("subWorkflowId")], + stop_when_same_turn=_reads_same_turn(execution.get("workflowDefinition") or {}), + ) + + +def fetch_execution(api_client: Any, execution_id: str) -> dict: + """The execution with its tasks, as the server's raw JSON (not the SDK's typed model).""" + return api_client.call_api( + "/workflow/{workflowId}", "GET", + path_params={"workflowId": execution_id}, + query_params=[("includeTasks", "true")], + header_params={"Accept": "application/json"}, + response_type="object", + auth_settings=[], + _return_http_data_only=True, + ) + + +def llm_tokens(execution: dict) -> Tuple[int, int]: + """Prompt and completion tokens over every LLM call in one execution (not its sub-workflows).""" + outputs = [t.get("outputData") or {} for t in execution.get("tasks") or [] + if t.get("taskType") == "LLM_CHAT_COMPLETE"] + return (sum(int(o.get("promptTokens") or 0) for o in outputs), + sum(int(o.get("completionTokens") or 0) for o in outputs)) + + +def _story_sent_to_critic(llm_output: dict) -> Optional[str]: + for call in llm_output.get("toolCalls") or []: + if call.get("name") == CRITIC_TOOL_NAME: + request = (call.get("inputParameters") or {}).get("request") + if isinstance(request, str): + return _LABEL.sub("", request, count=1).strip() + return None + + +def _workflow_result(execution: dict) -> str: + result = (execution.get("output") or {}).get("result") + return result.strip() if isinstance(result, str) else "" + + +def _reads_same_turn(definition: dict) -> Optional[bool]: + try: + find_stop_when(definition, STORYTELLER_NAME) + except RuntimeError: + return None + return stop_when_reads_same_turn(definition, STORYTELLER_NAME) diff --git a/harness/agentic/stop_wiring.py b/harness/agentic/stop_wiring.py new file mode 100644 index 000000000..f43b89992 --- /dev/null +++ b/harness/agentic/stop_wiring.py @@ -0,0 +1,102 @@ +"""Let the storyteller's stop_when see the critic's verdict on the turn the critic gives it. + +Why this hack exists: stop_when's decision depends on the result of a tool call, the critic's +verdict, but as compiled it can't see that result on the turn the tool call happens. We edit +the workflow so that it can. + +As compiled, stop_when's ``messages`` input is the storyteller LLM call's *input* messages, a +snapshot taken before that turn's critic call runs. The critic's verdict only reaches that +snapshot when the next turn's LLM input is assembled, so stop_when sees each verdict one turn +late and the loop runs one extra storyteller turn after the critic approves. +``_last_tool_results`` is a workflow variable the compiler already keeps: every tool result so +far, newest last, written after the critic and before stop_when. Pointing ``messages`` at it +lets stop_when see the verdict on the same turn and stop the loop on the approving turn. + +This edits the compiler's output, so it has to be re-applied after every deploy. The proper +fix is in the server's compiler: pass this turn's tool results to stop_when. +""" +from __future__ import annotations + +from typing import Any, Iterator + +SAME_TURN_MESSAGES = "${workflow.variables._last_tool_results}" +_TOOL_RESULTS_VARIABLE = "_last_tool_results" + + +def rewire_stop_when(api: Any, workflow_name: str, version: int = 1) -> bool: + """Point the stored definition's stop_when at this turn's tool results. + + ``api`` is a MetadataResourceApi. Works on the raw JSON, so fields the SDK's models don't + know about survive. Refuses to write a definition that lacks its agent markers, because + writing it back without them would make the server treat it as a plain workflow. Re-reads + the definition after writing to confirm. Returns True if it changed the definition, False + if it was already rewired. + """ + definition = _get(api, workflow_name, version) + _check_rewirable(definition, workflow_name, version) + stop = find_stop_when(definition, workflow_name) + if stop["inputParameters"].get("messages") == SAME_TURN_MESSAGES: + return False + stop["inputParameters"]["messages"] = SAME_TURN_MESSAGES + api.update1([definition], overwrite=True) + + stored = _get(api, workflow_name, version) + _check_rewirable(stored, workflow_name, version) + if find_stop_when(stored, workflow_name)["inputParameters"].get("messages") != SAME_TURN_MESSAGES: + raise RuntimeError(f"{workflow_name} v{version}: the rewired stop_when input did not persist") + return True + + +def stop_when_reads_same_turn(definition: dict, workflow_name: str) -> bool: + """True if this definition's stop_when has been rewired.""" + return find_stop_when(definition, workflow_name)["inputParameters"].get("messages") == SAME_TURN_MESSAGES + + +def find_stop_when(definition: dict, workflow_name: str) -> dict: + ref = f"{workflow_name}_stop_when" + for task in walk_tasks(definition.get("tasks") or []): + if task.get("taskReferenceName") == ref: + task.setdefault("inputParameters", {}) + return task + raise RuntimeError(f"{workflow_name} has no {ref} task") + + +def walk_tasks(tasks: list) -> Iterator[dict]: + """Every task in a definition, including those nested in loops, switches and forks.""" + for task in tasks: + yield task + yield from walk_tasks(task.get("loopOver") or []) + yield from walk_tasks(task.get("defaultCase") or []) + for branch in (task.get("decisionCases") or {}).values(): + yield from walk_tasks(branch) + for branch in task.get("forkTasks") or []: + yield from walk_tasks(branch) + + +def _get(api: Any, workflow_name: str, version: int) -> dict: + # Not api.get(): that parses into the SDK's WorkflowDef model, which can drop fields the + # model doesn't know about, and its raw mode (_preload_content=False) crashes on the + # SDK's httpx transport. response_type="object" returns the parsed JSON untouched. + return api.api_client.call_api( + "/metadata/workflow/{name}", "GET", + path_params={"name": workflow_name}, + query_params=[("version", version)], + header_params={"Accept": "application/json"}, + response_type="object", + auth_settings=[], + _return_http_data_only=True, + ) + + +def _check_rewirable(definition: dict, workflow_name: str, version: int) -> None: + metadata = definition.get("metadata") or {} + if not (metadata.get("agent_sdk") or metadata.get("agentDef")): + raise RuntimeError( + f"{workflow_name} v{version} has no agent markers in its metadata; not writing it back, " + "since the server would then treat it as a plain workflow") + find_stop_when(definition, workflow_name) + sets_variable = any( + task.get("type") == "SET_VARIABLE" and _TOOL_RESULTS_VARIABLE in (task.get("inputParameters") or {}) + for task in walk_tasks(definition.get("tasks") or [])) + if not sets_variable: + raise RuntimeError(f"{workflow_name} v{version} never sets {_TOOL_RESULTS_VARIABLE}") diff --git a/harness/try_agentic.py b/harness/try_agentic.py index 1ada5c71a..0a753a256 100644 --- a/harness/try_agentic.py +++ b/harness/try_agentic.py @@ -25,7 +25,11 @@ from conductor.ai.agents import AgentRuntime from conductor.client.configuration.configuration import Configuration +from conductor.client.http.api.metadata_resource_api import MetadataResourceApi +from conductor.client.http.api_client import ApiClient +from agentic.run_report import fetch_execution, llm_tokens, summarize +from agentic.stop_wiring import rewire_stop_when from agentic.story_agents import ( CRITIC_TOOL_NAME, STORYTELLER_NAME, @@ -55,6 +59,7 @@ def main() -> int: rng = random.Random(seed) concept = invent_concept(rng) configuration = Configuration() + args.configuration = configuration args.server = configuration.host print(f"server: {configuration.host}") @@ -91,6 +96,10 @@ def parse_args() -> argparse.Namespace: run.add_argument("--setting", help="story setting (default: random pick)") run.add_argument("--timeout", type=int, default=600, help="seconds before the run is cancelled (default: 600)") + run.add_argument("--stop-same-turn", action="store_true", + help="before the run, rewire stop_when to see the critic's verdict on the turn " + "the critic gives it, saving the extra storyteller call after approval " + "(edits the stored workflow; see agentic/stop_wiring.py)") run.add_argument("--quiet", action="store_true", help="no live progress, just the report") run.add_argument("--open", action="store_true", help="open the execution in your browser as soon as the run starts") @@ -180,6 +189,17 @@ def run_story(args: argparse.Namespace, runtime: AgentRuntime, rng: random.Rando # the SDK then starts worker processes only if _has_worker_tools() is true, which # ignores stop_when, so this agent's stop_when worker would never run. runtime.serve(agent, blocking=False) + if args.stop_same_turn: + # stop_when's decision depends on the critic's verdict, the result of a tool call, which + # as compiled it only sees one turn late. serve() just redeployed the definition, so + # the edit is re-applied here every time. See agentic/stop_wiring.py. + try: + changed = rewire_stop_when(MetadataResourceApi(ApiClient(args.configuration)), STORYTELLER_NAME) + except RuntimeError as e: + print(f"Not starting the run; could not rewire stop_when: {e}") + return 1 + print("stop_when: rewired to see the critic's verdict on the same turn" + + ("" if changed else " (it already was)") + "\n") handle = runtime.start(STORYTELLER_NAME, story_prompt(setting), version=1) link = _execution_url(args.ui_url or args.server, handle.execution_id) print(f"execution: {handle.execution_id}") @@ -214,7 +234,7 @@ def consume() -> None: print(f"\nInterrupted; cancelled {handle.execution_id}.") return 130 - return _report(handle, stream, progress, link) + return _report(args, handle, link) class _Progress: @@ -246,32 +266,64 @@ def _say(self, line: str) -> None: print(line if len(line) <= limit else line[:limit - 4] + " ...") -def _report(handle: Any, stream: Any, progress: _Progress, link: str) -> int: +def _report(args: argparse.Namespace, handle: Any, link: str) -> int: + """Report from the execution record. Streamed events can't be relied on: Orkes 5.6.0-rc4 + streams no tool-call events, so an event-based report misses every critic call.""" + api_client = ApiClient(args.configuration) try: - result = stream.get_result() - status = str(getattr(result.status, "value", result.status)) - output, usage, error = result.output, result.token_usage, result.error - except Exception: - snapshot = handle.get_status() - status, output, usage, error = snapshot.status, snapshot.output, None, snapshot.reason + summary = summarize(fetch_execution(api_client, handle.execution_id)) + except Exception as e: + print(f"\nCould not fetch the execution to report on it: {type(e).__name__}: {e}") + print(f"view: {link}") + return 1 + critic_tokens = _critic_tokens(api_client, summary.critic_execution_ids) - approved = any(v is not None and v.get("done") is True for v in progress.verdicts) print("\n" + "=" * WRAP) - print(f"status: {status}") - print(f"execution: {handle.execution_id}") - print(f"view: {link}") - print(f"critic calls: {len(progress.verdicts)} approved: " - f"{'yes' if approved else 'no (stopped at the turn limit, or failed)'}") - if usage is not None: - print(f"tokens: {usage.prompt_tokens} in, {usage.completion_tokens} out, " - f"{usage.total_tokens} total") - if error: - print(f"error: {error}") - - story = output if isinstance(output, str) and output.strip() else progress.story - print("\nstory:\n") - print(textwrap.fill(story, WRAP) if story else "(no story text came back)") - return 0 if status.upper() == "COMPLETED" else 1 + print(f"status: {summary.status}") + print(f"execution: {handle.execution_id}") + print(f"view: {link}") + print(f"stop_when: {_describe_wiring(summary.stop_when_same_turn)}") + print(f"approved: {f'yes, on turn {summary.approved_on}' if summary.approved_on else 'no'}") + tokens_in, tokens_out = summary.storyteller_tokens + tokens = f"storyteller {tokens_in:,} in / {tokens_out:,} out" + if critic_tokens is not None: + tokens += f"; critic {critic_tokens[0]:,} in / {critic_tokens[1]:,} out" + print(f"tokens: {tokens}") + + print("\nturns:") + for turn in summary.turns: + if turn.story is None: + what = "replied without calling the critic" if turn.finish_reason else "LLM call did not complete" + print(f" {turn.number:>2} ({what})") + else: + print(_clip(f" {turn.number:>2} + {turn.added}")) + print(_clip(f" critic: {_describe_verdict(turn.verdict)}")) + + print(f"\nstory ({summary.story_source}):\n") + print(textwrap.fill(summary.story, WRAP) if summary.story else "(no story)") + return 0 if summary.status.upper() == "COMPLETED" else 1 + + +def _critic_tokens(api_client: Any, critic_execution_ids: list) -> Optional[tuple]: + """Tokens over the critic's sub-workflows, or None if any can't be fetched.""" + try: + totals = [llm_tokens(fetch_execution(api_client, i)) for i in critic_execution_ids] + except Exception: + return None + return sum(t[0] for t in totals), sum(t[1] for t in totals) + + +def _describe_wiring(same_turn: Optional[bool]) -> str: + if same_turn is None: + return "unknown (no stop_when task found in the run's definition)" + if same_turn: + return "saw the critic's verdict on the same turn (rewired by --stop-same-turn)" + return "saw the critic's verdict one turn late (as compiled)" + + +def _clip(line: str) -> str: + limit = WRAP + 40 + return line if len(line) <= limit else line[:limit - 4] + " ..." def _describe_verdict(verdict: Optional[dict]) -> str: @@ -283,10 +335,10 @@ def _describe_verdict(verdict: Optional[dict]) -> str: def _execution_url(base: str, execution_id: str) -> str: - """UI link for an execution. Agent runs use the normal execution page in both the - OSS and Orkes UIs; the agent view is a tab on it. Printed as a plain URL, which - VS Code's terminal and iTerm2 make Cmd-clickable and which survives logs and - copy-paste.""" + """UI link for an execution. Agent runs use the normal execution page, which switches + to its agent view by itself when the workflow definition carries agent metadata. + Printed as a plain URL, which VS Code's terminal and iTerm2 make Cmd-clickable and + which survives logs and copy-paste.""" base = base.rstrip("/") if base.endswith("/api"): base = base[: -len("/api")] From 50c6518b682e31e020a647e8b14aef6f43cb71f9 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Mon, 2 Nov 2026 10:22:57 -0600 Subject: [PATCH 5/8] feat(harness): rewire stop_when in harness runs by default The harness governor now applies the same stop_when edit as try_agentic.py run --stop-same-turn, so stop_when sees the critic's verdict on the approving turn and each run saves one storyteller call. HARNESS_AGENTIC_STOP_SAME_TURN=false turns it off. - agentic/agentic_governor.py: every deploy undoes the edit, so it is due after the initial serve() and after each concept rotation, and is applied before the next run starts. If it fails (refused, HTTP or connection error), the governor logs a warning and does not start that run; it tries again once at the next interval, without retrying in a loop. - agentic/agentic_testing.py: reads the env var (default true) and hands the governor a MetadataResourceApi when it is on. - agentic/stop_wiring.py, README: why the edit reuses the messages input (the Python SDK's stop_when worker drops any other input), and the proper fix: single-agent stop_when can't see this turn's tool results; add a toolResults input in the server's compiler and expose it in each SDK's stop_when context. - README: describe running against Orkes only, where the provider key is stored in an AI integration and the model is /. Tested so far only with try_agentic.py against Orkes 5.6.0-rc4; drop the untested Conductor OSS instructions and the mock/mockLLM claim. Governor behavior checked with a fake runtime and a fake rewire; the harness mode has not been run on a cluster yet. Co-Authored-By: Claude Opus 5.5 --- harness/README.md | 37 ++++++++++++++++++----------- harness/agentic/agentic_governor.py | 31 +++++++++++++++++++++++- harness/agentic/agentic_testing.py | 10 ++++++-- harness/agentic/stop_wiring.py | 8 +++++-- 4 files changed, 67 insertions(+), 19 deletions(-) diff --git a/harness/README.md b/harness/README.md index 3e4eaf65f..832cba844 100644 --- a/harness/README.md +++ b/harness/README.md @@ -108,6 +108,8 @@ Off by default. When enabled, the harness also exercises the agent runtime. It c Each run adds one sentence per turn. After every sentence the storyteller calls a `python_story_critic` sub-agent, which returns a JSON verdict: whether the story meets the concept, and what is still missing. The `stop_when` worker ends the loop once the critic approves. If the critic never approves, `HARNESS_AGENTIC_MAX_TURNS` ends it. Either way the run completes. After `HARNESS_AGENTIC_RUNS_PER_CONCEPT` runs, the harness invents a new concept and redeploys over the same name, which replaces version 1 in place rather than adding versions. +By default the harness edits the deployed definition so that `stop_when` sees each verdict on the turn the critic gives it, the same edit as `try_agentic.py run --stop-same-turn` (see [below](#--stop-same-turn)). Every deploy undoes the edit, so the governor re-applies it after the first deploy and after each concept rotation, before it starts the next run. If the edit fails, the governor logs a warning, does not start that run, and tries again at the next interval. Set `HARNESS_AGENTIC_STOP_SAME_TURN=false` to run with the definition as compiled. The proper fix is described under [`--stop-same-turn`](#--stop-same-turn). + The worker logs each verdict it sees: ```text @@ -117,25 +119,29 @@ The worker logs each verdict it sees: ### Requirements -The LLM call runs on the Conductor server, so the provider key belongs in the **server's** environment, not the harness's. Conductor OSS enables the AI integration by default and reads `ANTHROPIC_API_KEY` into `conductor.ai.anthropic.api-key`. +So far this has been tested only with `try_agentic.py` against Orkes Conductor 5.6.0-rc4. + +The LLM calls run on the Conductor server, so the harness needs no provider key. The key is stored in an AI integration on the Orkes cluster: -### Running Locally Against Conductor OSS +- The cluster needs an AI integration for the provider, holding its API key, with the model enabled. +- `HARNESS_AGENTIC_MODEL` takes the form `/`. The default, `anthropic/claude-haiku-4-5`, works only if the integration is named `anthropic`. +- If the integration or the model is missing, the server fails the storyteller's first LLM task with a terminal error when it schedules it. + +### Running Against Orkes Conductor ```bash -# Terminal 1: start the server with the provider key in its environment -export ANTHROPIC_API_KEY=... -npm install -g @conductor-oss/conductor-cli -conductor server start +export CONDUCTOR_SERVER_URL=https://your-cluster.example.com/api +export CONDUCTOR_AUTH_KEY=$CONDUCTOR_AUTH_KEY +export CONDUCTOR_AUTH_SECRET=$CONDUCTOR_AUTH_SECRET -# Terminal 2: agentic testing only -export CONDUCTOR_SERVER_URL=http://localhost:8080/api +# Agentic testing only HARNESS_SDK_TESTING_ENABLED=false \ HARNESS_AGENTIC_TESTING_ENABLED=true \ -HARNESS_AGENTIC_INTERVAL_SECONDS=60 \ +HARNESS_AGENTIC_MODEL=/claude-haiku-4-5 \ python3 harness/main.py ``` -Open the execution in the Conductor UI to see each storyteller turn and critic call. +Open the execution in the Orkes UI to see each storyteller turn and critic call. ### Trying One Run @@ -166,23 +172,26 @@ A run is cancelled on Ctrl-C or after `--timeout` seconds (default 600), so it d Passing the agent object straight to `start()` doesn't work for this agent. The SDK then starts worker processes only if `_has_worker_tools()` is true, and that check ignores `stop_when`. The storyteller's only tool is the critic, which runs on the server, so its `stop_when` worker is registered but never started, and the run waits forever. -`run` registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. On Orkes, `--model` takes the AI integration's name before the slash. +`run` registers the agent as `python_storyteller`, the name the harness uses, so don't run it against a cluster where the harness's agentic testing is enabled. `--model` takes the AI integration's name before the slash, as `HARNESS_AGENTIC_MODEL` does. #### `--stop-same-turn` `stop_when` decides whether the loop ends, and that decision depends on the critic's verdict: the result of a tool call. As compiled, `stop_when` reads the storyteller LLM call's input messages, a snapshot taken before that turn's critic call, so it sees each verdict one turn late and the loop runs one extra storyteller call after the critic approves. That call carries the whole conversation, so it is the most expensive turn. -`--stop-same-turn` edits the stored workflow definition after `serve()` deploys it and before the run starts: it points `stop_when` at `_last_tool_results`, a workflow variable that already holds every tool result so far and is updated after the critic and before `stop_when`. The loop then stops on the approving turn. The edit is made on the raw JSON so no field is lost, it refuses to write a definition without its agent markers in `metadata`, and it re-reads the definition to confirm. Every deploy regenerates the definition, so the edit has to be re-applied each time; the report's `stop_when:` line says which wiring the run actually used. The proper fix is in the server's compiler: pass this turn's tool results to `stop_when`. See `agentic/stop_wiring.py`. +`--stop-same-turn` edits the stored workflow definition after `serve()` deploys it and before the run starts: it points `stop_when` at `_last_tool_results`, a workflow variable that already holds every tool result so far and is updated after the critic and before `stop_when`. The loop then stops on the approving turn. The edit is made on the raw JSON so no field is lost, it refuses to write a definition without its agent markers in `metadata`, and it re-reads the definition to confirm. Every deploy regenerates the definition, so the edit has to be re-applied each time; the report's `stop_when:` line says which wiring the run actually used. The harness applies the same edit by default; see `HARNESS_AGENTIC_STOP_SAME_TURN`. The edit reuses the `messages` input because the Python SDK's `stop_when` worker passes only `result`, `messages` and `iteration` to the function and drops any other input. See `agentic/stop_wiring.py`. + +The proper fix: single-agent `stop_when` can't see this turn's tool results; add a `toolResults` input in the server's compiler and expose it in each SDK's `stop_when` context. ### Agentic Environment Variables | Variable | Default | Description | |---|---|---| -| `HARNESS_AGENTIC_MODEL` | `anthropic/claude-haiku-4-5` | Model for the storyteller and the critic, in `provider/model` form | +| `HARNESS_AGENTIC_MODEL` | `anthropic/claude-haiku-4-5` | Model for the storyteller and the critic, in `/` form | | `HARNESS_AGENTIC_INTERVAL_SECONDS` | 300 | Seconds between run starts | | `HARNESS_AGENTIC_RUNS_PER_CONCEPT` | 3 | Runs per concept before a new one is invented and redeployed | | `HARNESS_AGENTIC_MAX_TURNS` | 20 | Turn limit for a run whose story the critic never approves | -| `HARNESS_AGENTIC_SEED` | random | Fixed seed for a reproducible concept and setting sequence. Required to replay recordings through `mock/mockLLM` | +| `HARNESS_AGENTIC_STOP_SAME_TURN` | true | Edit the deployed definition so `stop_when` sees the critic's verdict on the same turn. A run is not started while the edit fails | +| `HARNESS_AGENTIC_SEED` | random | Fixed seed for a reproducible concept and setting sequence | | `HARNESS_AGENTIC_DEBUG` | false | Print the raw message history when the worker finds no verdict | **Cost:** each run makes two LLM calls per turn, one for the storyteller and one for the critic. Because every turn resends the growing story, input tokens grow faster than linearly with turn count. On Claude Haiku 4.5 a 15-turn run costs roughly $0.04. At the default 300-second interval that is about $10 per day. Lower the cadence before you leave it running. diff --git a/harness/agentic/agentic_governor.py b/harness/agentic/agentic_governor.py index 378a4024f..131fb1b74 100644 --- a/harness/agentic/agentic_governor.py +++ b/harness/agentic/agentic_governor.py @@ -4,7 +4,9 @@ import threading from conductor.ai.agents import AgentRuntime +from conductor.client.http.api.metadata_resource_api import MetadataResourceApi +from agentic.stop_wiring import rewire_stop_when from agentic.story_agents import STORYTELLER_NAME, build_storyteller, story_prompt from agentic.story_concepts import StoryConcept, invent_concept, pick_setting @@ -18,6 +20,12 @@ class AgenticGovernor: overwrites the previous concept in place -- definitions never accumulate. Redeploying is plain HTTP and registers no workers, so it is safe from this thread; starting by name likewise skips worker registration. + + With a ``metadata_api``, it also rewires stop_when so it sees the critic's verdict on + the same turn (see agentic/stop_wiring.py). Every deploy regenerates the definition and + undoes the edit, so the edit is due after the initial deploy and after each rotation, + and is applied before the next run starts. If it fails, that run is not started and the + edit is tried again at the next interval. """ def __init__( @@ -29,6 +37,7 @@ def __init__( max_turns: int, interval_seconds: int, runs_per_concept: int, + metadata_api: MetadataResourceApi | None = None, ) -> None: self._runtime = runtime self._rng = rng @@ -38,13 +47,17 @@ def __init__( self._interval_seconds = interval_seconds self._runs_per_concept = runs_per_concept self._runs_left = runs_per_concept + self._metadata_api = metadata_api + # The caller deployed the first concept just before creating the governor. + self._rewire_due = metadata_api is not None self._stop_event = threading.Event() self._thread: threading.Thread | None = None def start(self) -> None: print( f"AgenticGovernor started: agent={STORYTELLER_NAME}, " - f"interval={self._interval_seconds}s, runs_per_concept={self._runs_per_concept}" + f"interval={self._interval_seconds}s, runs_per_concept={self._runs_per_concept}, " + f"stop_same_turn={self._metadata_api is not None}" ) self._stop_event.clear() self._thread = threading.Thread(target=self._run, name="AgenticGovernor", daemon=True) @@ -65,6 +78,8 @@ def _start_run(self) -> None: try: if self._runs_left <= 0: self._rotate_concept() + if self._rewire_due and not self._rewire_stop_when(): + return setting = pick_setting(self._rng) handle = self._runtime.start( STORYTELLER_NAME, story_prompt(setting), version=1, @@ -79,4 +94,18 @@ def _rotate_concept(self) -> None: self._runtime.deploy(build_storyteller(concept, self._model, self._max_turns)) self._concept = concept self._runs_left = self._runs_per_concept + self._rewire_due = self._metadata_api is not None print(f"AgenticGovernor: deployed new concept: {concept.summary()}") + + def _rewire_stop_when(self) -> bool: + """Apply the stop_when edit. Returns False, after logging why, if the run must not start.""" + try: + changed = rewire_stop_when(self._metadata_api, STORYTELLER_NAME) + except Exception as e: + print(f"AgenticGovernor: WARNING: could not rewire stop_when, not starting this run; " + f"retrying at the next interval: {e}") + return False + self._rewire_due = False + print("AgenticGovernor: rewired stop_when to see the critic's verdict on the same turn" + + ("" if changed else " (it already was)")) + return True diff --git a/harness/agentic/agentic_testing.py b/harness/agentic/agentic_testing.py index 24e8bea5f..40d4143ea 100644 --- a/harness/agentic/agentic_testing.py +++ b/harness/agentic/agentic_testing.py @@ -6,11 +6,13 @@ from conductor.ai.agents import AgentRuntime from conductor.client.configuration.configuration import Configuration +from conductor.client.http.api.metadata_resource_api import MetadataResourceApi +from conductor.client.http.api_client import ApiClient from agentic.agentic_governor import AgenticGovernor from agentic.story_agents import STORYTELLER_NAME, build_storyteller from agentic.story_concepts import invent_concept -from harness_env import env_int_or_default, env_str_or_default +from harness_env import env_bool_or_default, env_int_or_default, env_str_or_default class AgenticTesting: @@ -25,6 +27,9 @@ def __init__(self, configuration: Configuration) -> None: self._max_turns = env_int_or_default("HARNESS_AGENTIC_MAX_TURNS", 20) self._interval_seconds = env_int_or_default("HARNESS_AGENTIC_INTERVAL_SECONDS", 300) self._runs_per_concept = env_int_or_default("HARNESS_AGENTIC_RUNS_PER_CONCEPT", 3) + # Rewire stop_when so it sees the critic's verdict on the turn the critic gives it, + # saving one storyteller call per run. See agentic/stop_wiring.py. + self._stop_same_turn = env_bool_or_default("HARNESS_AGENTIC_STOP_SAME_TURN", True) # A fixed seed reproduces the same concept and setting sequence, which is what # recording once and replaying through mock/mockLLM requires. self._rng = random.Random(os.environ.get("HARNESS_AGENTIC_SEED") or None) @@ -41,9 +46,10 @@ def prepare(self) -> None: ) print(f"AgenticTesting: serving {STORYTELLER_NAME} on {self._model}, " f"first concept: {concept.summary()}") + metadata_api = MetadataResourceApi(ApiClient(self._configuration)) if self._stop_same_turn else None self._governor = AgenticGovernor( self._runtime, self._rng, concept, self._model, self._max_turns, - self._interval_seconds, self._runs_per_concept, + self._interval_seconds, self._runs_per_concept, metadata_api, ) def start(self) -> None: diff --git a/harness/agentic/stop_wiring.py b/harness/agentic/stop_wiring.py index f43b89992..8dbc52eb0 100644 --- a/harness/agentic/stop_wiring.py +++ b/harness/agentic/stop_wiring.py @@ -12,8 +12,12 @@ far, newest last, written after the critic and before stop_when. Pointing ``messages`` at it lets stop_when see the verdict on the same turn and stop the loop on the approving turn. -This edits the compiler's output, so it has to be re-applied after every deploy. The proper -fix is in the server's compiler: pass this turn's tool results to stop_when. +This edits the compiler's output, so it has to be re-applied after every deploy. It reuses +the ``messages`` input because the Python SDK's stop_when worker passes only ``result``, +``messages`` and ``iteration`` to the function and drops any other input. + +The proper fix: single-agent stop_when can't see this turn's tool results; add a +``toolResults`` input in the server's compiler and expose it in each SDK's stop_when context. """ from __future__ import annotations From 34d7076c20df0fb7489dc5e24327ec54a452141e Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Tue, 3 Nov 2026 09:35:30 -0600 Subject: [PATCH 6/8] feat(harness): report each agentic run's outcome and story The governor started runs and never looked at them again, so a run that failed (for example, a wrong integration name) was silent in the log. - agentic/agentic_governor.py: between starts, check the last run every 15 seconds and report it as soon as it has finished: status, turns, the turn the critic approved and whether the run stopped on it, which stop_when wiring it used, critic calls, storyteller tokens, a link, and the story the critic approved. Checks that find it still running are silent. If it hasn't finished by the next start, that start says so once and stops tracking it. On stop, name a run not yet reported. - The story is read from the execution, not the workflow output. A run that stops on the approving turn ends on a critic call, and the server's synth_output fallback then drops the tool-call inputs: its JSON.stringify serializes Java maps as {}. Reproduced with the server's GraalJSEvaluator. - agentic/run_report.py: add the failure reason; execution_url() moves here from try_agentic.py. - Remove --ui-url and HARNESS_UI_URL: links are always the server URL without /api, which is where Orkes serves the UI. - agentic/stop_wiring.py, README: name the proposed stop_when input tool_results, as the SWARM handoff checks already do. - README: explain why the concept and the setting are separate; add a Known Issues section for the server, SDK and CLI problems found so far; note testing on Orkes 5.6.0-rc4 and rc6 (both OSS 3.32.1). Checked against a run on Orkes 5.6.0-rc6 and with fakes. SDK testing code is unchanged. Co-Authored-By: Claude Opus 5.5 --- harness/README.md | 47 ++++++++++++++- harness/agentic/agentic_governor.py | 88 ++++++++++++++++++++++++++++- harness/agentic/agentic_testing.py | 7 ++- harness/agentic/run_report.py | 13 +++++ harness/agentic/stop_wiring.py | 4 +- harness/try_agentic.py | 18 +----- 6 files changed, 153 insertions(+), 24 deletions(-) diff --git a/harness/README.md b/harness/README.md index 832cba844..71ba783a8 100644 --- a/harness/README.md +++ b/harness/README.md @@ -110,6 +110,8 @@ Each run adds one sentence per turn. After every sentence the storyteller calls By default the harness edits the deployed definition so that `stop_when` sees each verdict on the turn the critic gives it, the same edit as `try_agentic.py run --stop-same-turn` (see [below](#--stop-same-turn)). Every deploy undoes the edit, so the governor re-applies it after the first deploy and after each concept rotation, before it starts the next run. If the edit fails, the governor logs a warning, does not start that run, and tries again at the next interval. Set `HARNESS_AGENTIC_STOP_SAME_TURN=false` to run with the definition as compiled. The proper fix is described under [`--stop-same-turn`](#--stop-same-turn). +The concept is compiled into the agent's instructions and the critic's rubric, so changing it means redeploying. The setting is only in each run's prompt, so it varies every run without a redeploy. + The worker logs each verdict it sees: ```text @@ -117,9 +119,25 @@ The worker logs each verdict it sees: [python_storyteller] turn 7: latest verdict done=True missing=[] ``` +Between starts, the governor checks the last run every 15 seconds and reports it as soon as it has finished, read from its execution record. The outcome is followed by the story: + +```text +AgenticGovernor: run COMPLETED after 10 turn(s): stopped on the approving turn 10; stop_when rewired; 10 critic call(s); storyteller tokens 27,163 in / 1,825 out https:///execution/ + story (approved by the critic on turn 10): + Twelve-year-old Maya clutched her canvas bag of lost keys as the ferry horn wailed through the + suffocating fog. ... +AgenticGovernor: run FAILED after 0 turn(s): No such model configured or enabled. integration: chatbotsxz, model: claude-haiku-4-5 https:///execution/ +``` + +The story is read from the execution, not the workflow output: a run that stops on the approving turn has no story in its output (see [Known Issues](#known-issues)). + +With the compiled wiring, a 6-turn story's run says `COMPLETED after 7 turn(s): approved on turn 6, stopped 1 turn(s) later; stop_when as compiled; 6 critic call(s); storyteller tokens 14,034 in / 1,089 out`: the extra turn cost 3,201 input tokens. + +Each run is reported once. Checks that find it still running are silent. If it hasn't finished by the next start, that start reports it as still running and stops checking it. + ### Requirements -So far this has been tested only with `try_agentic.py` against Orkes Conductor 5.6.0-rc4. +So far this has been tested only against Orkes Conductor 5.6.0-rc4 and 5.6.0-rc6, with both `try_agentic.py` and the harness. The LLM calls run on the Conductor server, so the harness needs no provider key. The key is stored in an AI integration on the Orkes cluster: @@ -164,7 +182,7 @@ python3 harness/try_agentic.py run --stop-same-turn The report is built from the execution record after the run, not from streamed events: Orkes 5.6.0-rc4 streams no tool-call events, so live per-turn lines appear only on servers that do. The story it prints is the text the critic approved, exactly as the critic judged it. The storyteller's own final reply can add commentary such as a checklist of the requirements. -`run` prints a link to the execution in the UI as soon as it starts (Cmd-click it in VS Code or iTerm2), and `--open` opens it in your browser. The link assumes the UI is served at the server URL without `/api`; if it isn't, pass `--ui-url` or set `HARNESS_UI_URL`. +`run` prints a link to the execution in the UI as soon as it starts (Cmd-click it in VS Code or iTerm2), and `--open` opens it in your browser. The link is the server URL without `/api`, which is where Orkes serves the UI. A run is cancelled on Ctrl-C or after `--timeout` seconds (default 600), so it doesn't keep calling the LLM. @@ -180,7 +198,30 @@ Passing the agent object straight to `start()` doesn't work for this agent. The `--stop-same-turn` edits the stored workflow definition after `serve()` deploys it and before the run starts: it points `stop_when` at `_last_tool_results`, a workflow variable that already holds every tool result so far and is updated after the critic and before `stop_when`. The loop then stops on the approving turn. The edit is made on the raw JSON so no field is lost, it refuses to write a definition without its agent markers in `metadata`, and it re-reads the definition to confirm. Every deploy regenerates the definition, so the edit has to be re-applied each time; the report's `stop_when:` line says which wiring the run actually used. The harness applies the same edit by default; see `HARNESS_AGENTIC_STOP_SAME_TURN`. The edit reuses the `messages` input because the Python SDK's `stop_when` worker passes only `result`, `messages` and `iteration` to the function and drops any other input. See `agentic/stop_wiring.py`. -The proper fix: single-agent `stop_when` can't see this turn's tool results; add a `toolResults` input in the server's compiler and expose it in each SDK's `stop_when` context. +The proper fix: single-agent `stop_when` can't see this turn's tool results; add a `tool_results` input in the server's compiler and expose it in each SDK's `stop_when` context. SWARM handoff checks already receive `${workflow.variables._last_tool_results}` under that name, so the server has precedent for this wiring. + +### Known Issues + +Problems found while building agentic testing, and how the harness works around them. Observed on Orkes Conductor 5.6.0-rc4 and 5.6.0-rc6, which both use Conductor OSS 3.32.1. + +**Server** + +- **`stop_when` sees tool results one turn late.** See [`--stop-same-turn`](#--stop-same-turn). Worked around by the stop_when edit, which is on by default. +- **A run that ends on a tool call loses its output.** When the loop stops on a turn where the LLM called a tool, the `synth_output` step builds the workflow output from that turn's tool calls, and their inputs come out empty: `[{"name":"story_critic","inputs":{}}]`. The step's script calls `JSON.stringify` on the inputs, which arrive as Java maps, and GraalJS serializes a Java map as `{}`. Reproduced with the server's own `GraalJSEvaluator`. With the stop_when edit every approved run ends this way, so the harness and `try_agentic.py` read the story from the execution record. +- **Orkes 5.6.0-rc4 streams no tool-call events**, so reports are built from the execution record. + +**Python SDK** + +- **`start(agent)` and `run(agent)` never start the `stop_when` worker** of an agent with no local tools: `_has_worker_tools()` ignores `stop_when`, so the run waits at `stop_when` forever. Workaround: `serve()`, then start by name. +- **The `stop_when` worker drops every task input except `result`, `messages` and `iteration`**, which is why the stop_when edit reuses `messages`. +- **`run(on_event=...)` ignores `timeout`.** `try_agentic.py` reads the event stream on a thread with its own deadline. +- **Raw responses (`_preload_content=False`) crash on the httpx transport**, which has `status_code` where the SDK reads `status`. Workaround: `call_api(..., response_type="object")`. +- **`examples/agents/kitchen_sink.py`'s `stop_when` predicate never stops the loop.** Its first parameter is `messages`, but it receives the context dict, so `messages[-1]` raises `KeyError`, which the worker logs and treats as "continue". +- Found while reading the code, not hit by the harness: `output_type` with a dataclass produces no schema, and `_type_to_json_schema(str | None)` returns `{}`. + +**Conductor CLI** + +- **It drops `metadata` from workflow definitions**, so round-tripping an agent definition through it would remove the agent markers and make the server treat it as a plain workflow. `agentic/stop_wiring.py` edits the raw JSON instead. ### Agentic Environment Variables diff --git a/harness/agentic/agentic_governor.py b/harness/agentic/agentic_governor.py index 131fb1b74..8c835482a 100644 --- a/harness/agentic/agentic_governor.py +++ b/harness/agentic/agentic_governor.py @@ -1,15 +1,23 @@ from __future__ import annotations import random +import textwrap import threading +import time from conductor.ai.agents import AgentRuntime from conductor.client.http.api.metadata_resource_api import MetadataResourceApi +from conductor.client.http.api_client import ApiClient +from agentic.run_report import RunSummary, execution_url, fetch_execution, summarize from agentic.stop_wiring import rewire_stop_when from agentic.story_agents import STORYTELLER_NAME, build_storyteller, story_prompt from agentic.story_concepts import StoryConcept, invent_concept, pick_setting +_TERMINAL_STATUSES = ("COMPLETED", "FAILED", "TERMINATED", "TIMED_OUT") +# How often the governor checks whether the last run has finished, between starts. +_POLL_SECONDS = 15 + class AgenticGovernor: """Starts one storyteller run per interval, each in a randomly picked setting. @@ -26,6 +34,11 @@ class AgenticGovernor: undoes the edit, so the edit is due after the initial deploy and after each rotation, and is applied before the next run starts. If it fails, that run is not started and the edit is tried again at the next interval. + + Between starts it checks the last run every ``_POLL_SECONDS`` and reports it as soon as it + has finished, read from its execution record: the outcome and the story. Checks that find + it still running are silent. If it hasn't finished by the next start, that start reports + it as still running and stops tracking it. Each run is reported once. """ def __init__( @@ -37,6 +50,8 @@ def __init__( max_turns: int, interval_seconds: int, runs_per_concept: int, + api_client: ApiClient, + server_url: str, metadata_api: MetadataResourceApi | None = None, ) -> None: self._runtime = runtime @@ -47,6 +62,9 @@ def __init__( self._interval_seconds = interval_seconds self._runs_per_concept = runs_per_concept self._runs_left = runs_per_concept + self._api_client = api_client + self._server_url = server_url + self._last_run: str | None = None self._metadata_api = metadata_api # The caller deployed the first concept just before creating the governor. self._rewire_due = metadata_api is not None @@ -67,14 +85,30 @@ def stop(self) -> None: self._stop_event.set() if self._thread is not None: self._thread.join(timeout=5.0) + if self._last_run is not None: + # Its stop_when worker stops with this process, so a run still in progress + # waits at stop_when until a harness serves it again. + print(f"AgenticGovernor: last run not yet reported, possibly still running: " + f"{self._last_run}") print("AgenticGovernor stopped") def _run(self) -> None: while not self._stop_event.is_set(): self._start_run() - self._stop_event.wait(timeout=self._interval_seconds) + self._wait_for_next_start() + + def _wait_for_next_start(self) -> None: + """Wait out the interval, reporting the last run as soon as a check finds it finished.""" + deadline = time.monotonic() + self._interval_seconds + while True: + remaining = deadline - time.monotonic() + if remaining <= 0 or self._stop_event.wait(timeout=min(_POLL_SECONDS, remaining)): + return + if self._last_run is not None: + self._report_last_run(final=False) def _start_run(self) -> None: + self._report_last_run() try: if self._runs_left <= 0: self._rotate_concept() @@ -85,10 +119,40 @@ def _start_run(self) -> None: STORYTELLER_NAME, story_prompt(setting), version=1, ) self._runs_left -= 1 - print(f"AgenticGovernor: started {handle.execution_id} (setting: {setting})") + self._last_run = handle.execution_id + print(f"AgenticGovernor: started {handle.execution_id} (setting: {setting}) " + f"{execution_url(self._server_url, handle.execution_id)}") except Exception as e: print(f"AgenticGovernor: error starting storyteller run: {e}") + def _report_last_run(self, final: bool = True) -> None: + """Report the last run and forget it if it has finished. Between starts (``final`` False) + say nothing otherwise, so it is checked again; before a start, also report a run that is + still going or can't be fetched, and forget it.""" + execution_id = self._last_run + if execution_id is None: + return + link = execution_url(self._server_url, execution_id) + try: + summary = summarize(fetch_execution(self._api_client, execution_id)) + except Exception as e: + if final: + self._last_run = None + print(f"AgenticGovernor: could not fetch run {execution_id} to report on it: {e} {link}") + return + if summary.status.upper() in _TERMINAL_STATUSES: + self._last_run = None + print(f"AgenticGovernor: run {execution_id} {_describe_outcome(summary)} {link}") + if summary.story: + # From the execution, not the workflow output: a run that stops on the approving + # turn ends on a critic call, so its output holds no story text. + print(f" story ({summary.story_source}):") + print(textwrap.fill(summary.story, 100, initial_indent=" ", subsequent_indent=" ")) + elif final: + self._last_run = None + print(f"AgenticGovernor: run {execution_id} still {summary.status} " + f"after {len(summary.turns)} turn(s); not reporting it again {link}") + def _rotate_concept(self) -> None: concept = invent_concept(self._rng) self._runtime.deploy(build_storyteller(concept, self._model, self._max_turns)) @@ -109,3 +173,23 @@ def _rewire_stop_when(self) -> bool: print("AgenticGovernor: rewired stop_when to see the critic's verdict on the same turn" + ("" if changed else " (it already was)")) return True + + +def _describe_outcome(summary: RunSummary) -> str: + """One line: status and turns, then the failure reason or the approval details.""" + turns = len(summary.turns) + outcome = f"{summary.status} after {turns} turn(s)" + if summary.status.upper() != "COMPLETED": + return f"{outcome}: {summary.reason or 'no reason given'}" + if summary.approved_on is None: + approval = "critic never approved" + elif summary.approved_on == summary.turns[-1].number: + approval = f"stopped on the approving turn {summary.approved_on}" + else: + extra = summary.turns[-1].number - summary.approved_on + approval = f"approved on turn {summary.approved_on}, stopped {extra} turn(s) later" + wiring = {True: "rewired", False: "as compiled", None: "unknown"}[summary.stop_when_same_turn] + tokens_in, tokens_out = summary.storyteller_tokens + return (f"{outcome}: {approval}; stop_when {wiring}; " + f"{len(summary.critic_execution_ids)} critic call(s); " + f"storyteller tokens {tokens_in:,} in / {tokens_out:,} out") diff --git a/harness/agentic/agentic_testing.py b/harness/agentic/agentic_testing.py index 40d4143ea..0a33e5d77 100644 --- a/harness/agentic/agentic_testing.py +++ b/harness/agentic/agentic_testing.py @@ -46,10 +46,13 @@ def prepare(self) -> None: ) print(f"AgenticTesting: serving {STORYTELLER_NAME} on {self._model}, " f"first concept: {concept.summary()}") - metadata_api = MetadataResourceApi(ApiClient(self._configuration)) if self._stop_same_turn else None + api_client = ApiClient(self._configuration) self._governor = AgenticGovernor( self._runtime, self._rng, concept, self._model, self._max_turns, - self._interval_seconds, self._runs_per_concept, metadata_api, + self._interval_seconds, self._runs_per_concept, + api_client=api_client, + server_url=self._configuration.host, + metadata_api=MetadataResourceApi(api_client) if self._stop_same_turn else None, ) def start(self) -> None: diff --git a/harness/agentic/run_report.py b/harness/agentic/run_report.py index 9f7b37bd2..ef3d4d53f 100644 --- a/harness/agentic/run_report.py +++ b/harness/agentic/run_report.py @@ -28,6 +28,7 @@ class Turn: @dataclass class RunSummary: status: str + reason: str # why the run failed or was terminated; empty if it didn't turns: List[Turn] approved_on: Optional[int] story: str @@ -68,6 +69,7 @@ def summarize(execution: dict) -> RunSummary: return RunSummary( status=str(execution.get("status") or "UNKNOWN"), + reason=str(execution.get("reasonForIncompletion") or ""), turns=turns, approved_on=approved.number if approved is not None else None, story=story, @@ -92,6 +94,17 @@ def fetch_execution(api_client: Any, execution_id: str) -> dict: ) +def execution_url(base: str, execution_id: str) -> str: + """UI link for an execution. Agent runs use the normal execution page, which switches + to its agent view by itself when the workflow definition carries agent metadata. + Printed as a plain URL, which VS Code's terminal and iTerm2 make Cmd-clickable and + which survives logs and copy-paste.""" + base = base.rstrip("/") + if base.endswith("/api"): + base = base[: -len("/api")] + return f"{base}/execution/{execution_id}" + + def llm_tokens(execution: dict) -> Tuple[int, int]: """Prompt and completion tokens over every LLM call in one execution (not its sub-workflows).""" outputs = [t.get("outputData") or {} for t in execution.get("tasks") or [] diff --git a/harness/agentic/stop_wiring.py b/harness/agentic/stop_wiring.py index 8dbc52eb0..61d66e0e3 100644 --- a/harness/agentic/stop_wiring.py +++ b/harness/agentic/stop_wiring.py @@ -17,7 +17,9 @@ ``messages`` and ``iteration`` to the function and drops any other input. The proper fix: single-agent stop_when can't see this turn's tool results; add a -``toolResults`` input in the server's compiler and expose it in each SDK's stop_when context. +``tool_results`` input in the server's compiler and expose it in each SDK's stop_when context. +SWARM handoff checks already receive ``${workflow.variables._last_tool_results}`` under that +name, so the server has precedent for this wiring. """ from __future__ import annotations diff --git a/harness/try_agentic.py b/harness/try_agentic.py index 0a753a256..fb970e974 100644 --- a/harness/try_agentic.py +++ b/harness/try_agentic.py @@ -28,7 +28,7 @@ from conductor.client.http.api.metadata_resource_api import MetadataResourceApi from conductor.client.http.api_client import ApiClient -from agentic.run_report import fetch_execution, llm_tokens, summarize +from agentic.run_report import execution_url, fetch_execution, llm_tokens, summarize from agentic.stop_wiring import rewire_stop_when from agentic.story_agents import ( CRITIC_TOOL_NAME, @@ -103,9 +103,6 @@ def parse_args() -> argparse.Namespace: run.add_argument("--quiet", action="store_true", help="no live progress, just the report") run.add_argument("--open", action="store_true", help="open the execution in your browser as soon as the run starts") - run.add_argument("--ui-url", default=os.environ.get("HARNESS_UI_URL"), - help="UI address for execution links (default: $HARNESS_UI_URL, else the " - "server URL without /api)") run.add_argument("--debug", action="store_true", help="the stop_when worker prints raw message history when it finds no verdict") run.set_defaults(handler=run_story) @@ -201,7 +198,7 @@ def run_story(args: argparse.Namespace, runtime: AgentRuntime, rng: random.Rando print("stop_when: rewired to see the critic's verdict on the same turn" + ("" if changed else " (it already was)") + "\n") handle = runtime.start(STORYTELLER_NAME, story_prompt(setting), version=1) - link = _execution_url(args.ui_url or args.server, handle.execution_id) + link = execution_url(args.server, handle.execution_id) print(f"execution: {handle.execution_id}") print(f"view: {link}") if args.open and not webbrowser.open(link): @@ -334,17 +331,6 @@ def _describe_verdict(verdict: Optional[dict]) -> str: return f"not done, missing: {verdict.get('missing')}" -def _execution_url(base: str, execution_id: str) -> str: - """UI link for an execution. Agent runs use the normal execution page, which switches - to its agent view by itself when the workflow definition carries agent metadata. - Printed as a plain URL, which VS Code's terminal and iTerm2 make Cmd-clickable and - which survives logs and copy-paste.""" - base = base.rstrip("/") - if base.endswith("/api"): - base = base[: -len("/api")] - return f"{base}/execution/{execution_id}" - - def _field(obj: Any, *names: str) -> Any: """Read a key from a dict response, or an attribute from a model object.""" for name in names: From ec14a29cc70663214edb5d0c871987ee9c8cb28b Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Tue, 3 Nov 2026 13:06:10 -0600 Subject: [PATCH 7/8] feat(harness): single-replica agentic deployment for certification-aws python-agent-harness-worker runs only the agentic module, next to the SDK harness deployment. One replica with the Recreate strategy: each pod runs its own agentic governor, and two would overwrite each other's python_storyteller and python_story_critic definitions mid-run. The model names the chatbots AI integration on certification-aws. PYTHONUNBUFFERED=1 so the stop_when worker's progress lines, printed from a spawned child process, reach the pod log. Co-Authored-By: Claude Opus 5.5 --- harness/manifests/deployment-agent-aws.yaml | 108 ++++++++++++++++++++ 1 file changed, 108 insertions(+) create mode 100644 harness/manifests/deployment-agent-aws.yaml diff --git a/harness/manifests/deployment-agent-aws.yaml b/harness/manifests/deployment-agent-aws.yaml new file mode 100644 index 000000000..ca370e6c1 --- /dev/null +++ b/harness/manifests/deployment-agent-aws.yaml @@ -0,0 +1,108 @@ +# Agentic testing only, certification-aws only. Runs next to deployment.yaml. +# One replica: each pod runs its own agentic governor, and two would overwrite each other's +# python_storyteller and python_story_critic definitions mid-run. +# Connection details come from the same ConfigMap and Secret as deployment.yaml. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: python-agent-harness-worker + # namespace: xxxxx # supply this in kubectl command + labels: + app: python-agent-harness-worker +spec: + replicas: 1 + strategy: + # Stop the old pod before starting the new one, so two governors never overlap. + type: Recreate + selector: + matchLabels: + app: python-agent-harness-worker + template: + metadata: + labels: + app: python-agent-harness-worker + spec: + containers: + - name: harness + image: ghcr.io/conductor-oss/python-sdk/harness-worker:harness-agentic-testing-latest + imagePullPolicy: Always + env: + # === CONDUCTOR CONNECTION (from per-cloud ConfigMap) === + + - name: CONDUCTOR_SERVER_URL + valueFrom: + configMapKeyRef: + name: python-sdk-harness-config + key: CONDUCTOR_SERVER_URL + + - name: CONDUCTOR_AUTH_KEY + valueFrom: + configMapKeyRef: + name: python-sdk-harness-config + key: CONDUCTOR_AUTH_KEY + + - name: CONDUCTOR_AUTH_SECRET + valueFrom: + secretKeyRef: + name: conductor-credentials + key: auth-secret + + # === AGENTIC TESTING ONLY === + - name: HARNESS_SDK_TESTING_ENABLED + value: "false" + + - name: HARNESS_AGENTIC_TESTING_ENABLED + value: "true" + + # / + - name: HARNESS_AGENTIC_MODEL + value: "chatbots/claude-haiku-4-5" + + - name: HARNESS_AGENTIC_INTERVAL_SECONDS + value: "300" + + # The stop_when worker runs in a spawned child process; without this its + # progress lines sit in a pipe buffer instead of reaching the pod log. + - name: PYTHONUNBUFFERED + value: "1" + + # === METRICS MODE === + - name: WORKER_CANONICAL_METRICS + value: "true" + + ports: + - name: metrics + containerPort: 9991 + protocol: TCP + + resources: + requests: + memory: "256Mi" + cpu: "100m" + limits: + memory: "512Mi" + cpu: "500m" + + livenessProbe: + exec: + command: + - /bin/sh + - -c + - test -e /proc/1/cmdline + initialDelaySeconds: 30 + periodSeconds: 30 + timeoutSeconds: 5 + failureThreshold: 3 + + readinessProbe: + exec: + command: + - /bin/sh + - -c + - test -e /proc/1/cmdline + initialDelaySeconds: 10 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 3 + + restartPolicy: Always From 5270da71bac98ad1d2578c6a0ca22efbc55fee75 Mon Sep 17 00:00:00 2001 From: Chris Hagglund Date: Tue, 3 Nov 2026 13:06:10 -0600 Subject: [PATCH 8/8] feat(harness): end agentic runs the harness can't finish correctly A run in progress when the agentic harness stops can't finish correctly: its stop_when worker is gone, and the next harness deploys a new concept, so the run calls a critic with different criteria from its next turn on (the critic is looked up by name on every call). - agentic/agentic_governor.py: on stop, report the last run if it has just finished, otherwise terminate it. - agentic/agentic_testing.py: on startup, before serve() deploys, terminate any python_storyteller run still RUNNING, e.g. one left by a crash. - agentic/story_agents.py: story_is_complete's docstring now describes both stop_when wirings. - README: describe both, and run only one agentic harness per cluster. Checked with fakes; not yet run on a cluster. Co-Authored-By: Claude Opus 5.5 --- harness/README.md | 2 ++ harness/agentic/agentic_governor.py | 38 ++++++++++++++++++++++++++--- harness/agentic/agentic_testing.py | 8 ++++-- harness/agentic/story_agents.py | 7 +++--- 4 files changed, 46 insertions(+), 9 deletions(-) diff --git a/harness/README.md b/harness/README.md index 71ba783a8..b144a9595 100644 --- a/harness/README.md +++ b/harness/README.md @@ -135,6 +135,8 @@ With the compiled wiring, a 6-turn story's run says `COMPLETED after 7 turn(s): Each run is reported once. Checks that find it still running are silent. If it hasn't finished by the next start, that start reports it as still running and stops checking it. +A run can't finish correctly once the harness that started it stops: its `stop_when` worker is gone, and the next harness deploys a new concept, which gives the run a different critic from its next turn on. So on shutdown the governor terminates its last run if it hasn't finished, and on startup, before deploying, the harness terminates any `python_storyteller` run still RUNNING (one left by a crash). Run only one agentic harness per cluster: each one deploys `python_storyteller` and `python_story_critic` with its own concept. + ### Requirements So far this has been tested only against Orkes Conductor 5.6.0-rc4 and 5.6.0-rc6, with both `try_agentic.py` and the harness. diff --git a/harness/agentic/agentic_governor.py b/harness/agentic/agentic_governor.py index 8c835482a..21649e387 100644 --- a/harness/agentic/agentic_governor.py +++ b/harness/agentic/agentic_governor.py @@ -7,6 +7,7 @@ from conductor.ai.agents import AgentRuntime from conductor.client.http.api.metadata_resource_api import MetadataResourceApi +from conductor.client.http.api.workflow_resource_api import WorkflowResourceApi from conductor.client.http.api_client import ApiClient from agentic.run_report import RunSummary, execution_url, fetch_execution, summarize @@ -39,6 +40,10 @@ class AgenticGovernor: has finished, read from its execution record: the outcome and the story. Checks that find it still running are silent. If it hasn't finished by the next start, that start reports it as still running and stops tracking it. Each run is reported once. + + On stop it terminates the last run if it hasn't finished. That run can't finish + correctly: its stop_when worker stops with this process, and the next harness to start + deploys a new concept, so the run would get a different critic from its next turn on. """ def __init__( @@ -63,6 +68,7 @@ def __init__( self._runs_per_concept = runs_per_concept self._runs_left = runs_per_concept self._api_client = api_client + self._workflow_api = WorkflowResourceApi(api_client) self._server_url = server_url self._last_run: str | None = None self._metadata_api = metadata_api @@ -85,11 +91,11 @@ def stop(self) -> None: self._stop_event.set() if self._thread is not None: self._thread.join(timeout=5.0) + # Report the last run if it has just finished; terminate it if it hasn't. + self._report_last_run(final=False) if self._last_run is not None: - # Its stop_when worker stops with this process, so a run still in progress - # waits at stop_when until a harness serves it again. - print(f"AgenticGovernor: last run not yet reported, possibly still running: " - f"{self._last_run}") + terminate_run(self._workflow_api, self._last_run, "agentic harness shutting down") + self._last_run = None print("AgenticGovernor stopped") def _run(self) -> None: @@ -175,6 +181,30 @@ def _rewire_stop_when(self) -> bool: return True +def terminate_running_runs(workflow_api: WorkflowResourceApi) -> None: + """Terminate every storyteller run still RUNNING, e.g. one left by a harness that crashed. + + Call before deploying: such a run would otherwise get the new concept's critic. + """ + try: + found = workflow_api.search( + query=f"workflowType IN ({STORYTELLER_NAME}) AND status IN (RUNNING)", size=100) + except Exception as e: + print(f"AgenticGovernor: could not search for leftover storyteller runs: {e}") + return + for summary in found.results or []: + terminate_run(workflow_api, summary.workflow_id, "left running by a previous agentic harness") + + +def terminate_run(workflow_api: WorkflowResourceApi, execution_id: str, reason: str) -> None: + try: + workflow_api.terminate(execution_id, reason=reason) + except Exception as e: + print(f"AgenticGovernor: could not terminate run {execution_id}: {e}") + return + print(f"AgenticGovernor: terminated run {execution_id} ({reason})") + + def _describe_outcome(summary: RunSummary) -> str: """One line: status and turns, then the failure reason or the approval details.""" turns = len(summary.turns) diff --git a/harness/agentic/agentic_testing.py b/harness/agentic/agentic_testing.py index 0a33e5d77..a3bcfec72 100644 --- a/harness/agentic/agentic_testing.py +++ b/harness/agentic/agentic_testing.py @@ -7,9 +7,10 @@ from conductor.ai.agents import AgentRuntime from conductor.client.configuration.configuration import Configuration from conductor.client.http.api.metadata_resource_api import MetadataResourceApi +from conductor.client.http.api.workflow_resource_api import WorkflowResourceApi from conductor.client.http.api_client import ApiClient -from agentic.agentic_governor import AgenticGovernor +from agentic.agentic_governor import AgenticGovernor, terminate_running_runs from agentic.story_agents import STORYTELLER_NAME, build_storyteller from agentic.story_concepts import invent_concept from harness_env import env_bool_or_default, env_int_or_default, env_str_or_default @@ -40,13 +41,16 @@ def __init__(self, configuration: Configuration) -> None: def prepare(self) -> None: """Deploy the first concept and fork the storyteller's worker processes. Starts no threads.""" self._runtime = AgentRuntime(configuration=self._configuration) + api_client = ApiClient(self._configuration) + # serve() deploys a new concept, and a run left by a previous harness would get its + # critic from the next turn on, so end any such run first. + terminate_running_runs(WorkflowResourceApi(api_client)) concept = invent_concept(self._rng) self._runtime.serve( build_storyteller(concept, self._model, self._max_turns), blocking=False, ) print(f"AgenticTesting: serving {STORYTELLER_NAME} on {self._model}, " f"first concept: {concept.summary()}") - api_client = ApiClient(self._configuration) self._governor = AgenticGovernor( self._runtime, self._rng, concept, self._model, self._max_turns, self._interval_seconds, self._runs_per_concept, diff --git a/harness/agentic/story_agents.py b/harness/agentic/story_agents.py index 70217ca0b..a5282a19d 100644 --- a/harness/agentic/story_agents.py +++ b/harness/agentic/story_agents.py @@ -89,9 +89,10 @@ def story_is_complete(context: dict) -> bool: """Loop exit gate: True once the critic's latest verdict says the story is done. Runs as a worker in a spawned process, so it must stay a module-level function: - spawn pickling cannot carry closures. ``context["messages"]`` is the LLM's *input* - history, so the newest verdict it can see is the previous turn's -- the loop ends - at most one turn after the critic approves. + spawn pickling cannot carry closures. As compiled, ``context["messages"]`` is the LLM's + *input* history, so the newest verdict it can see is the previous turn's and the loop + ends one turn after the critic approves. With the stop_when edit (agentic/stop_wiring.py) + it holds every tool result so far, so it sees the verdict on the turn the critic gives it. """ iteration = context.get("iteration", 0) verdict = find_verdict(context.get("messages"))