diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py index 6bf02cd6..41dc5035 100644 --- a/src/forge/workflow/nodes/implementation.py +++ b/src/forge/workflow/nodes/implementation.py @@ -108,12 +108,25 @@ async def implement_task(state: WorkflowState) -> WorkflowState: settings = get_settings() jira = JiraClient(settings) + # Initialize task_summary to handle early failures before Jira fetch + task_summary = "unknown" + try: # Get Task details from Jira task_issue = await jira.get_issue(current_task) task_description = task_issue.description or "" task_summary = task_issue.summary + logger.info( + f"Implementation started for task {current_task}", + extra={ + "event": "implementation_started", + "task_name": task_summary, + "feature_id": ticket_key, + "task_id": current_task, + }, + ) + # Post status comment at task implementation start await post_status_comment( jira, @@ -154,6 +167,16 @@ async def implement_task(state: WorkflowState) -> WorkflowState: if result.success: logger.info(f"Container completed successfully for {current_task}") + logger.info( + f"Implementation completed for task {current_task}", + extra={ + "event": "implementation_completed", + "task_name": task_summary, + "feature_id": ticket_key, + "task_id": current_task, + "success": True, + }, + ) # Post status comment at task implementation completion await post_status_comment( @@ -186,6 +209,17 @@ async def implement_task(state: WorkflowState) -> WorkflowState: except Exception as e: logger.error(f"Implementation failed for {current_task}: {e}") + # Emit structured end log for failed implementation + logger.info( + f"Implementation ended for task {current_task or 'unknown'}", + extra={ + "event": "implementation_ended", + "task_name": task_summary, + "feature_id": ticket_key, + "task_id": current_task or "unknown", + "success": False, + }, + ) return { **state, "last_error": str(e), diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index 64aeac11..1d97f2de 100644 --- a/tests/unit/workflow/nodes/test_implementation.py +++ b/tests/unit/workflow/nodes/test_implementation.py @@ -1,5 +1,6 @@ """Unit tests for implement_task node.""" +import logging from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -56,7 +57,6 @@ def _make_successful_runner(): class TestImplementTaskStartedComment: - @pytest.mark.asyncio async def test_posts_comment_on_task_ticket_before_container(self): """A comment is posted on the task ticket (not parent) when implementation starts.""" @@ -184,7 +184,6 @@ async def test_passes_trace_context_to_container_runner(self): class TestImplementationNodeRouting: - @pytest.mark.asyncio async def test_feature_missing_workspace_uses_feature_implementation_node(self): """Feature implementation failures must resume at implement_task.""" @@ -270,3 +269,623 @@ async def test_bug_container_failure_keeps_bug_implementation_node(self): assert result["current_node"] == "implement_bug_fix" assert result["last_error"] == "container failed" assert result["retry_count"] == 1 + + +class TestImplementationStartedStructuredLog: + """Tests for the implementation_started structured log event.""" + + @pytest.mark.asyncio + async def test_implementation_started_log_includes_structured_fields(self, caplog): + """Should emit structured log with event, task_name, feature_id, task_id.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Fix null pointer in AuthService") + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-123", + current_task_key="TASK-456", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation started")] + assert len(records) == 1 + record = records[0] + assert record.event == "implementation_started" + assert record.task_name == "Fix null pointer in AuthService" + assert record.feature_id == "FEAT-123" + assert record.task_id == "TASK-456" + + @pytest.mark.asyncio + async def test_implementation_started_log_empty_task_summary(self, caplog): + """Empty task summary logs with empty string for task_name.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="") + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task(_make_state()) + + records = [r for r in caplog.records if r.message.startswith("Implementation started")] + assert len(records) == 1 + assert records[0].task_name == "" + + @pytest.mark.asyncio + async def test_implementation_started_log_special_characters_in_summary(self, caplog): + """Special characters in task summary are logged as-is.""" + from forge.workflow.nodes.implementation import implement_task + + special_summary = "Fix bug with 'quotes' & and \"double quotes\"" + mock_jira = _make_mock_jira(summary=special_summary) + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task(_make_state()) + + records = [r for r in caplog.records if r.message.startswith("Implementation started")] + assert len(records) == 1 + assert records[0].task_name == special_summary + + +class TestImplementationCompletedStructuredLog: + """Tests for the implementation_completed structured log event.""" + + @pytest.mark.asyncio + async def test_implementation_completed_log_includes_structured_fields(self, caplog): + """Should emit structured log with event, task_name, feature_id, task_id, success.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Fix null pointer in AuthService") + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-123", + current_task_key="TASK-456", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation completed")] + assert len(records) == 1 + record = records[0] + assert record.event == "implementation_completed" + assert record.task_name == "Fix null pointer in AuthService" + assert record.feature_id == "FEAT-123" + assert record.task_id == "TASK-456" + assert record.success is True + + @pytest.mark.asyncio + async def test_implementation_completed_log_empty_task_summary(self, caplog): + """Empty task summary logs with empty string for task_name.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="") + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task(_make_state()) + + records = [r for r in caplog.records if r.message.startswith("Implementation completed")] + assert len(records) == 1 + assert records[0].task_name == "" + + @pytest.mark.asyncio + async def test_implementation_completed_log_special_characters_in_summary(self, caplog): + """Special characters in task summary are logged as-is.""" + from forge.workflow.nodes.implementation import implement_task + + special_summary = "Fix bug with 'quotes' & and \"double quotes\"" + mock_jira = _make_mock_jira(summary=special_summary) + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task(_make_state()) + + records = [r for r in caplog.records if r.message.startswith("Implementation completed")] + assert len(records) == 1 + assert records[0].task_name == special_summary + + @pytest.mark.asyncio + async def test_implementation_completed_log_not_emitted_on_failure(self, caplog): + """Should NOT emit implementation_completed log when container fails.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Fix null pointer in AuthService") + runner = MagicMock() + container_result = MagicMock() + container_result.success = False + container_result.error_message = "container failed" + runner.run = AsyncMock(return_value=container_result) + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task(_make_state()) + + records = [r for r in caplog.records if r.message.startswith("Implementation completed")] + assert len(records) == 0 + + +class TestImplementationEndedStructuredLog: + """Tests for the implementation_ended structured log event on failure.""" + + @pytest.mark.asyncio + async def test_implementation_ended_log_emitted_on_failure(self, caplog): + """Should emit structured log with event='implementation_ended' when container fails.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Fix null pointer in AuthService") + runner = MagicMock() + container_result = MagicMock() + container_result.success = False + container_result.error_message = "container failed" + runner.run = AsyncMock(return_value=container_result) + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task(_make_state()) + + records = [r for r in caplog.records if r.message.startswith("Implementation ended")] + assert len(records) == 1 + assert records[0].event == "implementation_ended" + + @pytest.mark.asyncio + async def test_implementation_ended_log_includes_structured_fields(self, caplog): + """Should emit structured log with task_name, feature_id, task_id, success=False.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Fix null pointer in AuthService") + runner = MagicMock() + container_result = MagicMock() + container_result.success = False + container_result.error_message = "container failed" + runner.run = AsyncMock(return_value=container_result) + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-123", + current_task_key="TASK-456", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation ended")] + assert len(records) == 1 + record = records[0] + assert record.task_name == "Fix null pointer in AuthService" + assert record.feature_id == "FEAT-123" + assert record.task_id == "TASK-456" + assert record.success is False + + @pytest.mark.asyncio + async def test_implementation_ended_log_uses_unknown_placeholder_for_task_summary(self, caplog): + """Should use 'unknown' for task_name when Jira fetch fails before setting task_summary.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = AsyncMock() + mock_jira.get_issue = AsyncMock(side_effect=Exception("Jira connection failed")) + mock_jira.close = AsyncMock() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-123", + current_task_key="TASK-456", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation ended")] + assert len(records) == 1 + record = records[0] + assert record.task_name == "unknown" + assert record.feature_id == "FEAT-123" + assert record.task_id == "TASK-456" + assert record.success is False + + @pytest.mark.asyncio + async def test_implementation_ended_log_uses_unknown_placeholder_for_task_id(self, caplog): + """Should use 'unknown' for task_id when current_task is None. + + Note: In practice, the exception handler is only reached when current_task + is set. This test verifies that the defensive `current_task or 'unknown'` + pattern works correctly by simulating what the code would produce. + """ + from forge.workflow.nodes import implementation + + # Simulate a None task_id being passed through the placeholder pattern + # This tests the same logic used in the exception handler: `current_task or "unknown"` + simulated_task_id: str | None = None + resolved_task_id = simulated_task_id or "unknown" + + with caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"): + implementation.logger.info( + f"Implementation ended for task {resolved_task_id}", + extra={ + "event": "implementation_ended", + "task_name": "Test Task", + "feature_id": "FEAT-123", + "task_id": resolved_task_id, + "success": False, + }, + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation ended")] + assert len(records) == 1 + record = records[0] + # Verify the placeholder is used in both message and structured field + assert "unknown" in record.message + assert record.task_id == "unknown" + + +class TestImplementTaskStructuredLogging: + """Consolidated tests for structured logging in implement_task (TS-001 through TS-007).""" + + @pytest.mark.asyncio + async def test_logs_implementation_started_with_structured_fields(self, caplog): + """TS-001: Start log has correct extra fields (event, task_name, feature_id, task_id).""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add new validation logic") + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-100", + current_task_key="TASK-200", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation started")] + assert len(records) == 1 + record = records[0] + assert record.levelno == logging.INFO + assert record.event == "implementation_started" + assert record.task_name == "Add new validation logic" + assert record.feature_id == "FEAT-100" + assert record.task_id == "TASK-200" + + @pytest.mark.asyncio + async def test_logs_implementation_completed_on_success(self, caplog): + """TS-002: Success end log has success=True.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Implement API endpoint") + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-101", + current_task_key="TASK-201", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation completed")] + assert len(records) == 1 + record = records[0] + assert record.levelno == logging.INFO + assert record.event == "implementation_completed" + assert record.task_name == "Implement API endpoint" + assert record.feature_id == "FEAT-101" + assert record.task_id == "TASK-201" + assert record.success is True + + @pytest.mark.asyncio + async def test_logs_implementation_ended_on_failure(self, caplog): + """TS-003: Failure end log has success=False when container fails.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Refactor database layer") + runner = MagicMock() + container_result = MagicMock() + container_result.success = False + container_result.error_message = "Tests failed in container" + runner.run = AsyncMock(return_value=container_result) + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-102", + current_task_key="TASK-202", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation ended")] + assert len(records) == 1 + record = records[0] + assert record.levelno == logging.INFO + assert record.event == "implementation_ended" + assert record.task_name == "Refactor database layer" + assert record.feature_id == "FEAT-102" + assert record.task_id == "TASK-202" + assert record.success is False + + @pytest.mark.asyncio + async def test_logs_implementation_ended_on_exception(self, caplog): + """TS-004: Exception path emits end log with success=False.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Handle edge case") + runner = MagicMock() + runner.run = AsyncMock(side_effect=RuntimeError("Unexpected container crash")) + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-103", + current_task_key="TASK-203", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation ended")] + assert len(records) == 1 + record = records[0] + assert record.levelno == logging.INFO + assert record.event == "implementation_ended" + assert record.task_name == "Handle edge case" + assert record.feature_id == "FEAT-103" + assert record.task_id == "TASK-203" + assert record.success is False + + @pytest.mark.asyncio + async def test_logs_unknown_task_name_when_jira_fails_early(self, caplog): + """TS-005: Placeholder 'unknown' used when Jira fetch fails before setting task_summary.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = AsyncMock() + mock_jira.get_issue = AsyncMock(side_effect=Exception("Jira unreachable")) + mock_jira.close = AsyncMock() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-104", + current_task_key="TASK-204", + ) + ) + + records = [r for r in caplog.records if r.message.startswith("Implementation ended")] + assert len(records) == 1 + record = records[0] + assert record.levelno == logging.INFO + assert record.event == "implementation_ended" + assert record.task_name == "unknown" + assert record.feature_id == "FEAT-104" + assert record.task_id == "TASK-204" + assert record.success is False + + @pytest.mark.asyncio + async def test_logs_empty_task_summary_as_empty_string(self, caplog): + """TS-006: Empty task summary logged as empty string.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="") + runner = _make_successful_runner() + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch( + "forge.workflow.nodes.implementation.ContainerRunner", + return_value=runner, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level(logging.INFO, logger="forge.workflow.nodes.implementation"), + ): + await implement_task(_make_state()) + + # Check start log + start_records = [ + r for r in caplog.records if r.message.startswith("Implementation started") + ] + assert len(start_records) == 1 + assert start_records[0].levelno == logging.INFO + assert start_records[0].task_name == "" + + # Check completion log + completed_records = [ + r for r in caplog.records if r.message.startswith("Implementation completed") + ] + assert len(completed_records) == 1 + assert completed_records[0].levelno == logging.INFO + assert completed_records[0].task_name == "" + + @pytest.mark.asyncio + async def test_logs_special_characters_in_task_summary(self, caplog): + """TS-007: Special characters in task summary are not escaped.""" + from forge.workflow.nodes.implementation import implement_task + + special_summary = "Fix