diff --git a/harness/README.md b/harness/README.md index 01346f4bb..b144a9595 100644 --- a/harness/README.md +++ b/harness/README.md @@ -99,3 +99,142 @@ 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. + +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 +[python_storyteller] turn 3: latest verdict done=False missing=['introduce the conflict'] +[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. + +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. + +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: + +- 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 +export CONDUCTOR_SERVER_URL=https://your-cluster.example.com/api +export CONDUCTOR_AUTH_KEY=$CONDUCTOR_AUTH_KEY +export CONDUCTOR_AUTH_SECRET=$CONDUCTOR_AUTH_SECRET + +# Agentic testing only +HARNESS_SDK_TESTING_ENABLED=false \ +HARNESS_AGENTIC_TESTING_ENABLED=true \ +HARNESS_AGENTIC_MODEL=/claude-haiku-4-5 \ + python3 harness/main.py +``` + +Open the execution in the Orkes 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, 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 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. + +`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. + +`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 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 `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 + +| Variable | Default | Description | +|---|---|---| +| `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_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/__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..21649e387 --- /dev/null +++ b/harness/agentic/agentic_governor.py @@ -0,0 +1,225 @@ +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.workflow_resource_api import WorkflowResourceApi +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. + + 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. + + 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. + + 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. + + 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__( + self, + runtime: AgentRuntime, + rng: random.Random, + concept: StoryConcept, + model: str, + 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 + 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._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 + # 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"stop_same_turn={self._metadata_api is not None}" + ) + 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) + # 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: + terminate_run(self._workflow_api, self._last_run, "agentic harness shutting down") + self._last_run = None + print("AgenticGovernor stopped") + + def _run(self) -> None: + while not self._stop_event.is_set(): + self._start_run() + 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() + 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, + ) + self._runs_left -= 1 + 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)) + 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 + + +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) + 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 new file mode 100644 index 000000000..a3bcfec72 --- /dev/null +++ b/harness/agentic/agentic_testing.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import os +import random +import threading + +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, 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 + + +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) + # 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) + 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) + 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()}") + self._governor = AgenticGovernor( + self._runtime, self._rng, concept, self._model, self._max_turns, + 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: + """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/run_report.py b/harness/agentic/run_report.py new file mode 100644 index 000000000..ef3d4d53f --- /dev/null +++ b/harness/agentic/run_report.py @@ -0,0 +1,135 @@ +"""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 + reason: str # why the run failed or was terminated; empty if it didn't + 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"), + reason=str(execution.get("reasonForIncompletion") or ""), + 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 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 [] + 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..61d66e0e3 --- /dev/null +++ b/harness/agentic/stop_wiring.py @@ -0,0 +1,108 @@ +"""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. 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 +``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 + +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/agentic/story_agents.py b/harness/agentic/story_agents.py new file mode 100644 index 000000000..a5282a19d --- /dev/null +++ b/harness/agentic/story_agents.py @@ -0,0 +1,154 @@ +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.""" + # 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: + 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. 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")) + 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/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 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..fb970e974 --- /dev/null +++ b/harness/try_agentic.py @@ -0,0 +1,346 @@ +"""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 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 +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 execution_url, fetch_execution, llm_tokens, summarize +from agentic.stop_wiring import rewire_stop_when +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 + +DEFAULT_MODEL = "anthropic/claude-haiku-4-5" +WRAP = 88 + + +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" + + seed = args.seed or secrets.token_hex(4) + rng = random.Random(seed) + concept = invent_concept(rng) + configuration = Configuration() + args.configuration = configuration + args.server = configuration.host + + 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)") + 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") + 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("--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") + 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) + # 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) + 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.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. + 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(args, handle, link) + + +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(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: + 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) + + print("\n" + "=" * WRAP) + 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: + 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())