diff --git a/packages/opencode-agents/README.md b/packages/opencode-agents/README.md index f186df8a..bcbdc2c1 100644 --- a/packages/opencode-agents/README.md +++ b/packages/opencode-agents/README.md @@ -48,4 +48,6 @@ asyncio.run(main()) `events` is required. Each invocation calls `events(config, user_input, variables, history, session_id)` and the return value must be a fresh async iterable of that session only. The iterable must end: return it after `session.idle`, or close it yourself. The handler yields AG-UI text and tool events. `config().stream()` yields `chunk` then `done`. -Pass `skip_judges=True`. This handler matches every agent-mode provider, including a judge, and it does not run the judge prompt. +Pass `skip_judges=True`. This handler matches every agent-mode provider, so judge selection can land here. It does not run the judge prompt: that call raises without calling `events`, so the judge is tracked as a failed generation and logged as an error. + +Do not pass this handler to `run_judge`. It would be picked as the agent fallback and `run_judge` would raise. diff --git a/packages/opencode-agents/agents.md b/packages/opencode-agents/agents.md index 2410195e..e47fecdd 100644 --- a/packages/opencode-agents/agents.md +++ b/packages/opencode-agents/agents.md @@ -38,6 +38,6 @@ Assistant text is emitted only after `message.updated` marks that message `role The AI Config, user input, variables, and history are arguments to `events`. A callable that ignores them never sees the prompt. -`provides_for` is `('*', 'agent')`, so a judge is routed here too. Pass `skip_judges=True` until this handler runs the prompt it is given. Otherwise `events` runs again on the judge prompt and the score is wrong. +`provides_for` is `('*', 'agent')`, so judge selection still lands here. Pass `skip_judges=True`. The handler does not run that prompt. A judge evaluation raises without calling `events`, so the judge is tracked as a failed generation. Do not pass this handler to `run_judge`, which would raise. `session_id` is required. The reducer does not adopt the first session on a shared stream. Other sessions are dropped, including their `session.error`. `session.idle` for this session ends the run. If the iterable is a server-wide stream and you neither send `session.idle` nor close it, `done` never arrives. diff --git a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py index a0aca9ce..f535f766 100644 --- a/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py +++ b/packages/opencode-agents/src/launchdarkly_ai_opencode_agents/handler.py @@ -11,6 +11,8 @@ from collections.abc import AsyncGenerator, AsyncIterable, Callable, Mapping from typing import Any +from opentelemetry import context as otel_context + from launchdarkly_ai_server import ( AiConfigRep, ProviderHandler, @@ -19,12 +21,15 @@ _config, _create_handler, add_cached_tokens_to_input, + compose_history, + content_to_text, end_span_once, set_input_content_attributes, set_output_content_attributes, text_message, ) from launchdarkly_ai_server.agui import TurnEmitter +from launchdarkly_ai_server.conversation import _EVAL_KEY from launchdarkly_ai_server.lifecycle import effective_capture_content from launchdarkly_ai_server.sdk_usage import report_usage from launchdarkly_ai_server.utils import number_or_zero @@ -460,20 +465,43 @@ def _session_error_message(properties: object) -> str: return f"The OpenCode session failed ({name})." +def _is_judge_evaluation() -> bool: + """Whether this call is a judge evaluation, not the agent's own session. + + ``provides_for`` stays ``('*', 'agent')``, so judge selection still lands + here. The judge prompt is not this OpenCode session. Calling *events* for + it would answer the judge with that session. + """ + return otel_context.get_value(_EVAL_KEY) is not None + + def _input_messages( user_input: str | None, history: list[dict[str, Any]] | None, ) -> list[SpanMessage]: + """History and *user_input* as span messages. + + Block-array history is flattened the same way as the other handlers. + System turns stay off this list: they belong on the system instructions. + An empty *user_input* adds nothing, so a history turn that already holds + the user message is not followed by a blank one. + """ + items = [item for item in (history or []) if isinstance(item, dict)] messages: list[SpanMessage] = [] - for item in history or []: - if not isinstance(item, dict): - continue + for item in compose_history(history=items, user_input=user_input): role = item.get("role") content = item.get("content") - if isinstance(role, str) and role and isinstance(content, str): + if not isinstance(role, str) or not role: + continue + if isinstance(content, str): messages.append(text_message(role, content)) - if isinstance(user_input, str): - messages.append(text_message("user", user_input)) + continue + if isinstance(content, list): + text = content_to_text( + [block for block in content if isinstance(block, dict)] + ) + if text: + messages.append(text_message(role, text)) return messages @@ -509,9 +537,11 @@ def create_opencode_handler( The stream yields the AG-UI text and tool subset and one ``done``. ``config().stream()`` still yields ``chunk`` then ``done``. Pass - ``skip_judges=True`` to ``config()``: a judge would otherwise be routed - here and *events* would run again on the judge prompt. *capture_content* - follows the other handlers. + ``skip_judges=True`` to ``config()``. Judge selection lands here, because + ``provides_for`` is ``('*', 'agent')``, and the handler does not run the + judge prompt: the call raises without calling *events*, so the judge is + tracked as a failed generation. Do not pass this handler to ``run_judge``. + *capture_content* follows the other handlers. """ report_usage("opencode-agents.createOpenCodeHandler", PACKAGE_NAME, __version__) return _create_opencode_handler( @@ -531,8 +561,9 @@ def opencode_agents( *events* is required and is the same callable :func:`create_opencode_handler` takes. *variables*, *history*, *session_id*, and *capture_content* configure - that call. Pass ``skip_judges=True`` so a judged config does not run - *events* again. + that call. Pass ``skip_judges=True``. A judge evaluation raises without + calling *events* again, so a judged config tracks the judge as a failed + generation. """ report_usage("opencode-agents.opencodeAgents", PACKAGE_NAME, __version__) variables = kwargs.pop("variables", None) @@ -573,6 +604,8 @@ async def _stream_impl( variables: dict[str, Any] | None, history: list[dict[str, Any]] | None = None, ) -> AsyncGenerator[dict[str, Any], None]: + if _is_judge_evaluation(): + raise RuntimeError("The OpenCode handler does not run judge prompts.") capture = effective_capture_content(content_override) span = start_root_span(config, variables or {}) ended: set[int] = set() diff --git a/packages/opencode-agents/tests/test_handler.py b/packages/opencode-agents/tests/test_handler.py index 4e806c23..e7084af6 100644 --- a/packages/opencode-agents/tests/test_handler.py +++ b/packages/opencode-agents/tests/test_handler.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from collections.abc import AsyncIterator, Iterator from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -149,6 +150,15 @@ def mock_ld_client() -> Iterator[MagicMock]: lifecycle_module._reset_for_testing() +def judge_errors(client: MagicMock) -> int: + return sum( + 1 + for call in client.track.call_args_list + if call.args[0] == "$ld:ai:generation:error" + and call.args[2].get("configKey") == "judge-flag" + ) + + class TestSurface: def test_provides_for_is_any_provider_in_agent_mode(self) -> None: handler = create_opencode_handler(events=source([]), session_id="ses") @@ -917,12 +927,20 @@ async def events( ] assert [event["type"] for event in skipped] == ["chunk", "done"] assert calls == ["q"] + assert not judge_errors(mock_ld_client) calls.clear() - await _drain(config(key="flag", handler=handler).stream("q", CONTEXT)) - assert calls[0] == "q" - assert len(calls) == 2 - assert calls[1] != "q" + judged_run = [ + event + async for event in config(key="flag", handler=handler).stream("q", CONTEXT) + ] + assert calls == ["q"] + assert judged_run[-1]["type"] == "done" + assert judged_run[-1]["response"] == "Hi" + assert not judged_run[-1].get("judge_results") + # Without skip_judges the raise is the safety net: the judge is + # tracked as a failed generation instead of scoring this session. + assert judge_errors(mock_ld_client) == 1 async def test_the_wrapper_forwards_variables_history_and_session( self, mock_ld_client: MagicMock @@ -1028,6 +1046,109 @@ def set_span_in_context(self, span: _Span) -> Any: assert tracer.spans[0].attributes["gen_ai.operation.name"] == "invoke_agent" assert tracer.spans[0].attributes["launchdarkly.operation.type"] == "gen_ai" + async def test_block_array_history_reaches_the_span( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + from launchdarkly_ai_opencode_agents import spans as span_module + + class _Span: + def __init__(self, name: str, context: Any = None) -> None: + self.name = name + self.context = context + self.attributes: dict[str, Any] = {} + self.events: list[str] = [] + + def set_attribute(self, key: str, value: Any) -> None: + self.attributes[key] = value + + def add_event( + self, name: str, attributes: dict[str, Any] | None = None + ) -> None: + self.events.append(name) + + def set_status(self, code: Any, description: str | None = None) -> None: + return None + + def record_exception(self, exc: BaseException) -> None: + return None + + def end(self) -> None: + return None + + class _Tracer: + def __init__(self) -> None: + self.spans: list[_Span] = [] + + def get_tracer(self, name: str) -> _Tracer: + return self + + def start_span(self, name: str, context: Any = None) -> _Span: + span = _Span(name, context) + self.spans.append(span) + return span + + def set_span_in_context(self, span: _Span) -> Any: + return span + + tracer = _Tracer() + monkeypatch.setattr(span_module, "trace", tracer) + handler = create_opencode_handler( + capture_content=True, + session_id="ses", + events=source( + [ + assistant_open("m1"), + text_delta("m1", "p1", "Hi"), + assistant_completed("m1"), + ] + ), + ) + history = [ + {"role": "system", "content": "do not record"}, + {"role": "user", "content": "earlier"}, + { + "role": "user", + "content": [ + {"type": "text", "text": "see "}, + { + "type": "image", + "source": { + "type": "url", + "url": "https://example.test/a.png", + }, + }, + {"type": "text", "text": "this"}, + ], + }, + { + "role": "assistant", + "content": [{"type": "text", "text": "noted"}], + }, + ] + seen = [ + event async for event in await handler.stream({}, "", None, None, history) + ] + assert seen[-1]["output"] == "Hi" + recorded = json.loads(tracer.spans[0].attributes["gen_ai.input.messages"]) + assert recorded == [ + { + "role": "user", + "parts": [{"type": "text", "content": "earlier"}], + }, + { + "role": "user", + "parts": [{"type": "text", "content": "see this"}], + }, + { + "role": "assistant", + "parts": [{"type": "text", "content": "noted"}], + }, + ] + assert ( + "do not record" not in tracer.spans[0].attributes["gen_ai.input.messages"] + ) + assert "gen_ai.prompt.3.role" not in tracer.spans[0].attributes + async def _drain(stream: Any) -> None: async for _event in stream: