from __future__ import annotations from unittest.mock import MagicMock, patch from behave import given, then, when from cleveragents.domain.models.core import Change, Context, Plan, Project from cleveragents.domain.models.core.change import OperationType from cleveragents.domain.providers.ai_provider import ProviderResponse from cleveragents.providers.llm.langchain_chat_provider import LangChainChatProvider @given("a LangChain chat provider is configured with a fake LangChain graph") def step_configure_langchain_provider(context): context.progress_updates = [] def progress_callback(percent: int) -> None: context.progress_updates.append(percent) context.progress_callback = progress_callback context.requested_models = [] context.llm_instance = MagicMock() context.llm_instance.get_num_tokens = MagicMock(return_value=0) def fake_llm_factory(model_id: str): context.requested_models.append(model_id) return context.llm_instance def build_provider(*, supports_streaming: bool) -> LangChainChatProvider: return LangChainChatProvider( name="test-langchain-provider", model_id="test-model", llm_factory=fake_llm_factory, max_retries=2, supports_streaming=supports_streaming, ) context.create_provider = build_provider context.provider = build_provider(supports_streaming=False) context.project = MagicMock(spec=Project) context.plan = MagicMock(spec=Plan) context.plan.prompt = "Implement feature" context.contexts = [MagicMock(spec=Context)] context.contexts[0].content = "Initial context" @when("the provider generates changes with a validation failure response") def step_provider_generates_validation_failure(context): generated_change = Change( plan_id=1, file_path="src/demo.py", operation=OperationType.MODIFY, new_content="print('hi')", ) validation_message = "Graph validation rejected the output" fake_state = { "generated_changes": [generated_change], "validation_result": {"status": "FAIL", "message": validation_message}, "error": None, } with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = fake_state context.graph_instance = mock_graph context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) constructor_args, constructor_kwargs = graph_cls.call_args context.graph_constructor_args = constructor_args context.graph_constructor_kwargs = constructor_kwargs context.expected_change = generated_change context.expected_validation_message = validation_message @when("the provider generates changes and the graph raises an exception") def step_provider_generates_exception(context): failure_message = "LangChain graph execution failed" with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.side_effect = RuntimeError(failure_message) try: context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) except RuntimeError as exc: # pragma: no cover context.unexpected_exception = exc context.graph_instance = mock_graph context.failure_message = failure_message @when("the provider generates changes and retries exhaust with nested errors") def step_provider_nested_retry_failure(context): context.failure_message = "Underlying LangChain failure" retry_error_payload = { "message": "Retries exhausted after 2 attempts", "error": { "message": context.failure_message, }, } fake_state = { "generated_changes": [], "validation_result": {}, "error": { "retry_error": retry_error_payload, "message": "RetryError", }, } with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = fake_state context.graph_instance = mock_graph context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) @when("the provider generates changes with opaque nested error containers") def step_provider_generates_with_opaque_nested_errors(context): nested_error = [ None, { "wrapper": { "layers": [ {"unused": None}, {"payload": {"metadata": {"message": "Nested failure message"}}}, ] } }, ] fake_state = { "generated_changes": [], "validation_result": {}, "error": nested_error, } with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = fake_state context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.failure_message = "Nested failure message" @when("the provider generates changes with overly deep error containers") def step_provider_generates_with_overly_deep_errors(context): def _build_deep_value(depth: int) -> dict[str, object]: value: dict[str, object] = {"detail": "Deep failure sentinel"} for _ in range(depth): value = {"layer": value} return value deep_error = {"labyrinth": _build_deep_value(12)} fake_state = { "generated_changes": [], "validation_result": {}, "error": deep_error, } with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = fake_state context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.deep_error_fragment = "Deep failure sentinel" @when("the provider generates changes with exception based errors") def step_provider_generates_with_exception_errors(context): failure_message = "Exception based failure" fake_state = { "generated_changes": [], "validation_result": {}, "error": RuntimeError(failure_message), } with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = fake_state context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.failure_message = failure_message @when("the provider generates changes with opaque error mappings") def step_provider_generates_with_opaque_error_mappings(context): opaque_error = {"mystery": None} fake_state = { "generated_changes": [], "validation_result": {}, "error": opaque_error, } with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = fake_state context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.expected_error_mapping = str(opaque_error) @when("the provider generates changes with numeric error payloads") def step_provider_generates_with_numeric_error_payloads(context): fake_state = { "generated_changes": [], "validation_result": {}, "error": 404, } with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = fake_state context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.failure_message = "404" @when( "the provider generates changes without a progress callback and the graph returns invalid response data" ) def step_provider_invalid_state_without_progress(context): context.llm_instance.get_num_tokens = None context.contexts = [MagicMock(spec=Context)] context.contexts[0].content = "" invalid_state = { "generated_changes": "not-a-list", "validation_result": "not-a-dict", "error": "Graph returned invalid payload", } context.state_error_message = invalid_state["error"] with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.invoke.return_value = invalid_state context.graph_instance = mock_graph context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=None, ) @when("the provider streams changes with incremental graph events") def step_provider_streams_changes(context): context.provider = context.create_provider(supports_streaming=True) generated_change = Change( plan_id=7, file_path="src/streamed.py", operation=OperationType.CREATE, new_content="print('streaming')", ) validation_message = "Looks great" streaming_events = [ {"load_context": {}}, {"analyze_requirements": {}}, {"generate_plan": {"generated_changes": [generated_change]}}, { "validate": { "validation_result": {"status": "PASS", "message": validation_message} } }, ] context.llm_instance.get_num_tokens.return_value = 321 with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.stream.return_value = iter(streaming_events) context.graph_instance = mock_graph context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) constructor_args, constructor_kwargs = graph_cls.call_args stream_call_args = mock_graph.stream.call_args context.graph_constructor_args = constructor_args context.graph_constructor_kwargs = constructor_kwargs context.stream_call_args = stream_call_args context.expected_change = generated_change context.expected_token_count = 321 @when("the provider stream API emits workflow events with fallback token estimation") def step_provider_stream_api_with_fallback_tokens(context): context.provider = context.create_provider(supports_streaming=True) context.llm_instance.get_num_tokens = MagicMock(return_value=99) generated_change = Change( plan_id=11, file_path="src/stream_api.py", operation=OperationType.MODIFY, new_content="print('api-stream')", ) streaming_events = [ {"__start__": {}}, {"load_context": {}}, {"analyze_requirements": {}}, {"generate_plan": {"generated_changes": [generated_change]}}, {"validate": {"validation_result": {"status": "PASS", "message": "OK"}}}, {"__end__": {}}, ] class _UsageContext: def __init__(self): self.tracker = MagicMock() self.tracker.total_tokens = {"invalid": "value"} self.tracker.total_cost = None def __enter__(self): return self.tracker def __exit__(self, exc_type, exc, tb): return False usage_context = _UsageContext() with ( patch.object( context.provider, "_usage_tracker", MagicMock(return_value=usage_context), ), patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls, ): mock_graph = graph_cls.return_value mock_graph.stream.return_value = iter(streaming_events) stream = context.provider.stream_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.stream_events = list(stream) context.stream_call_args = mock_graph.stream.call_args context.graph_constructor_args, context.graph_constructor_kwargs = ( graph_cls.call_args ) final_event = context.stream_events[-1] context.response = final_event["__end__"]["response"] context.expected_change = generated_change context.expected_token_count = 99 @when("the provider stream API emits non-dict payloads without a progress callback") def step_provider_stream_api_without_progress_non_dict(context): context.provider = context.create_provider(supports_streaming=True) context.llm_instance.get_num_tokens = MagicMock(return_value=7) context.contexts = [MagicMock(spec=Context)] context.contexts[0].content = "" streaming_events = [ {"load_context": "string-payload"}, {"analyze_requirements": {}}, {"generate_plan": {"generated_changes": []}}, {"validate": None}, ] with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.stream.return_value = iter(streaming_events) stream = context.provider.stream_changes( context.project, context.plan, context.contexts, progress_callback=None, ) context.stream_events = list(stream) context.stream_call_args = mock_graph.stream.call_args context.graph_constructor_args, context.graph_constructor_kwargs = ( graph_cls.call_args ) final_event = context.stream_events[-1]["__end__"] context.response = final_event["response"] context.expected_token_count = 7 context.non_dict_payload = streaming_events[0]["load_context"] @when( "the provider streams changes and the stream raises an exception after unknown node events" ) def step_provider_stream_failure_with_unknown_nodes(context): context.provider = context.create_provider(supports_streaming=True) failure_message = "Streaming interrupted by upstream failure" streaming_events = [{"load_context": {}}, {"mystery_node": "unexpected-payload"}] context.llm_instance.get_num_tokens.return_value = None def _failing_stream(): for event in streaming_events: yield event raise RuntimeError(failure_message) with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.stream.return_value = _failing_stream() context.graph_instance = mock_graph context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.failure_message = failure_message @when("the provider streams changes without streaming support") def step_provider_streams_without_support(context): fallback_change = Change( plan_id=21, file_path="src/fallback.py", operation=OperationType.CREATE, new_content="print('fallback')", ) fallback_response = ProviderResponse( changes=[fallback_change], model_used="test-model", token_count=7, error_message=None, ) with patch.object( context.provider, "generate_changes", MagicMock(return_value=fallback_response), ) as generate_changes_mock: stream = context.provider.stream_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.stream_events = list(stream) context.response = fallback_response context.fallback_response = fallback_response context.fallback_generate = generate_changes_mock @when("the provider stream API raises an error after partial events") def step_provider_stream_api_partial_failure(context): context.provider = context.create_provider(supports_streaming=True) context.llm_instance.get_num_tokens = MagicMock(return_value=0) failure_message = "Stream API failure after partial events" class _UsageContext: def __enter__(self): return None def __exit__(self, exc_type, exc, tb): return False def _partial_failure_stream(): yield {"__start__": {}} yield {"load_context": {}} yield {"analyze_requirements": {}} raise RuntimeError(failure_message) with ( patch.object( context.provider, "_usage_tracker", MagicMock(return_value=_UsageContext()), ), patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls, ): mock_graph = graph_cls.return_value mock_graph.stream.return_value = _partial_failure_stream() stream = context.provider.stream_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.stream_events = [] try: for event in stream: context.stream_events.append(event) except RuntimeError as exc: context.stream_failure = exc context.failure_message = failure_message @when("the provider stream API raises an error without a progress callback") def step_provider_stream_api_error_without_progress(context): context.provider = context.create_provider(supports_streaming=True) context.llm_instance.get_num_tokens = MagicMock(return_value=0) failure_message = "Stream API failure without progress callback" def _failing_stream(): yield {"load_context": {}} yield {"analyze_requirements": {}} raise RuntimeError(failure_message) with patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls: mock_graph = graph_cls.return_value mock_graph.stream.return_value = _failing_stream() stream = context.provider.stream_changes( context.project, context.plan, context.contexts, progress_callback=None, ) context.stream_events = [] try: for event in stream: context.stream_events.append(event) except RuntimeError as exc: context.stream_failure = exc context.failure_message = failure_message @when("the provider generates changes with an OpenAI usage callback") def step_provider_generates_with_openai_usage_callback(context): context.llm_instance.__class__.__module__ = "langchain_openai.chat_models" tracker = MagicMock() tracker.total_tokens = "84" tracker.total_cost = "1.25" context.expected_token_count = 84 context.expected_cost = 1.25 context.callback_invocations = 0 context.callback_entered = False context.callback_exited = False class _FakeUsageContext: def __enter__(self): context.callback_entered = True return tracker def __exit__(self, exc_type, exc, tb): context.callback_exited = True return False def fake_usage_callback(): context.callback_invocations += 1 return _FakeUsageContext() with ( patch( "cleveragents.providers.llm.langchain_chat_provider._openai_callback", fake_usage_callback, ), patch( "cleveragents.providers.llm.langchain_chat_provider.PlanGenerationGraph" ) as graph_cls, ): mock_graph = graph_cls.return_value mock_graph.invoke.return_value = { "generated_changes": [], "validation_result": {}, "error": None, } context.provider._logger = MagicMock() context.response = context.provider.generate_changes( context.project, context.plan, context.contexts, progress_callback=context.progress_callback, ) context.openai_tracker = tracker @when("the provider logs usage without token or cost data") def step_provider_logs_usage_without_metrics(context): context.provider._logger = MagicMock() context.provider._log_usage(model="test-model", tokens=None, cost=None) @then( "the provider response should contain the generated change data and validation error" ) def step_assert_response_contains_validation_error(context): assert context.response is not None assert context.response.changes, "Expected generated changes to be returned" assert context.response.changes[0].file_path == context.expected_change.file_path assert context.response.error_message == context.expected_validation_message, ( "Validation failure message should be surfaced" ) assert context.response.model_used == "test-model" assert context.response.token_count == 0 assert context.requested_models == ["test-model"] @then("the provider response should surface the state error without changes") def step_assert_state_error_without_changes(context): assert context.response is not None assert context.response.changes == [] assert context.response.error_message == context.state_error_message assert context.response.token_count == 0 @then("the progress callback should not receive updates") def step_assert_no_progress_updates(context): assert context.progress_updates == [] @then("the progress callback should record the LangChain workflow milestones") def step_assert_progress_milestones(context): assert context.progress_updates == [5, 90, 100] @then("the graph should be invoked with the supplied project context") def step_assert_graph_invocation(context): assert context.graph_instance.invoke.called, "Graph invoke should be called" call_args, call_kwargs = context.graph_instance.invoke.call_args assert call_args == ( context.project, context.plan, context.contexts, ) assert "thread_id" in call_kwargs assert call_kwargs["thread_id"].startswith("provider-") assert context.graph_constructor_args == () assert context.graph_constructor_kwargs["llm"] is context.llm_instance assert context.graph_constructor_kwargs["max_retries"] == 2 @then("the provider response should contain the streamed change payload") def step_assert_streaming_response(context): assert context.response is not None assert context.response.changes, "Expected streaming change results" streamed_change = context.response.changes[0] assert streamed_change.file_path == context.expected_change.file_path assert streamed_change.operation == context.expected_change.operation assert streamed_change.new_content == context.expected_change.new_content assert context.response.error_message in (None, "") @then("the stream API should emit workflow events with the provider response") def step_assert_stream_api_response(context): assert context.stream_events, "Expected stream API events" node_sequence = [next(iter(event.keys())) for event in context.stream_events] assert node_sequence[:-1] == [ "load_context", "analyze_requirements", "generate_plan", "validate", ] assert node_sequence[-1] == "__end__" final_event = context.stream_events[-1]["__end__"] assert isinstance(final_event["response"], ProviderResponse) assert final_event["response"] is context.response @then("the stream API should propagate the streaming error to the caller") def step_assert_stream_api_error(context): assert hasattr(context, "stream_failure"), "Expected streaming failure to surface" assert isinstance(context.stream_failure, RuntimeError) assert str(context.stream_failure) == context.failure_message node_sequence = [next(iter(event.keys())) for event in context.stream_events] assert node_sequence == ["load_context", "analyze_requirements"] @then("the stream API progress should include the final failure signal") def step_assert_stream_api_failure_progress(context): assert context.progress_updates == [5, 15, 40, 100] @then("the progress callback should include streaming milestones") def step_assert_streaming_progress(context): assert context.progress_updates == [5, 15, 40, 70, 90, 100] @then("the progress callback should capture partial streaming progress before failure") def step_assert_streaming_partial_progress(context): assert context.progress_updates == [5, 15, 100, 100] @then("the LangChain graph stream should receive the thread-aware configuration") def step_assert_stream_call(context): call_args, call_kwargs = context.stream_call_args assert call_args == ( context.project, context.plan, context.contexts, ) assert "thread_id" in call_kwargs assert call_kwargs["thread_id"].startswith("provider-") @then("the provider should report the estimated token usage") def step_assert_token_count(context): assert context.response is not None assert context.response.token_count == context.expected_token_count @then("the fallback stream should replay workflow milestones with the response") def step_assert_fallback_stream(context): assert context.fallback_generate.call_count == 1 nodes = [next(iter(event.keys())) for event in context.stream_events] assert nodes == [ "load_context", "analyze_requirements", "generate_plan", "validate", "__end__", ] generate_plan_payload = context.stream_events[2]["generate_plan"] assert generate_plan_payload["status"] == "completed" assert generate_plan_payload["change_count"] == len( context.fallback_response.changes ) final_event = context.stream_events[-1] assert final_event["__end__"]["response"] is context.fallback_response @then("the provider should report the callback-derived usage metrics") def step_assert_openai_usage_metrics(context): assert context.callback_invocations == 1 assert context.callback_entered is True assert context.callback_exited is True assert context.response.token_count == context.expected_token_count assert context.provider._logger.info.called _, logged_kwargs = context.provider._logger.info.call_args assert logged_kwargs["tokens"] == context.expected_token_count assert logged_kwargs["cost"] == context.expected_cost @then("the provider should skip emitting usage metrics") def step_assert_usage_logging_skipped(context): context.provider._logger.info.assert_not_called() @then("the provider response should capture the graph failure") def step_assert_graph_failure_response(context): assert context.response is not None assert context.response.changes == [] assert context.response.error_message == context.failure_message assert context.response.model_used == "test-model" assert context.response.token_count == 0 @then("the provider response should capture the nested retry failure") def step_assert_nested_retry_failure(context): assert context.response is not None, "Expected provider response to be populated" assert context.response.changes == [], ( f"Expected no changes for nested failure, got {context.response.changes!r}" ) assert context.response.error_message == context.failure_message, ( f"Expected error message {context.failure_message!r}, got {context.response.error_message!r}" ) @then("the progress callback should end at 100 percent even on failure") def step_assert_failure_progress_updates(context): assert context.progress_updates, "Expected progress updates to be recorded" assert context.progress_updates[0] == 5, ( f"Expected first progress update to be 5, got {context.progress_updates[0]!r}" ) assert context.progress_updates[-1] == 100, ( f"Expected final progress update to be 100, got {context.progress_updates[-1]!r}" ) @then("the stream API should preserve non-dict payloads in events") def step_assert_stream_preserves_non_dict_payloads(context): assert context.stream_events, "Expected stream API events" first_event = context.stream_events[0] assert "load_context" in first_event assert first_event["load_context"] == context.non_dict_payload @then("the provider response should expose the opaque nested error message") def step_assert_opaque_nested_error_message(context): assert context.response is not None assert context.response.changes == [] assert context.response.error_message == context.failure_message @then("the provider response should include the truncated deep error context") def step_assert_deep_error_context(context): assert context.response is not None assert context.deep_error_fragment in (context.response.error_message or "") @then("the provider response should stringify the opaque error mapping") def step_assert_stringified_error_mapping(context): assert context.response is not None assert context.response.error_message == context.expected_error_mapping