From fe2b4d44d59f3cc6c786070022ea7bdaa90a9b46 Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 16:48:21 +0000 Subject: [PATCH 1/8] [AISOS-2201] Add structured start log to implement_task function Added INFO-level structured log message at the beginning of task implementation: - Log emitted after Jira issue fetch with event="implementation_started" - Includes task_name, feature_id, task_id in extra dict - Empty task summary logs as empty string - Special characters in task summary logged as-is Closes: AISOS-2201 --- src/forge/workflow/nodes/implementation.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py index 5514cc26..1fd7c0b4 100644 --- a/src/forge/workflow/nodes/implementation.py +++ b/src/forge/workflow/nodes/implementation.py @@ -172,6 +172,16 @@ async def implement_task(state: WorkflowState) -> WorkflowState: 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 or "", + "feature_id": ticket_key, + "task_id": current_task, + }, + ) + # Post status comment at task implementation start await post_status_comment( jira, From c6f650f13e810b169cd4a05e2fc7bfec9c94ab3b Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 16:51:39 +0000 Subject: [PATCH 2/8] [AISOS-2201] Add unit tests for structured implementation_started log Added test class TestImplementationStartedStructuredLog with three tests: - test_implementation_started_log_emitted_with_extra_fields: verifies INFO log with event, task_name, feature_id, task_id in extra dict - test_implementation_started_log_empty_summary: verifies empty summary logs as task_name="" - test_implementation_started_log_special_characters_preserved: verifies special characters in task summary are logged as-is Uses pytest caplog fixture to capture and verify log records with extra fields. Closes: AISOS-2201 --- .../workflow/nodes/test_implementation.py | 119 ++++++++++++++++++ 1 file changed, 119 insertions(+) diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index 90de8ba8..cdf71377 100644 --- a/tests/unit/workflow/nodes/test_implementation.py +++ b/tests/unit/workflow/nodes/test_implementation.py @@ -370,3 +370,122 @@ 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 structured log emitted when implementation starts.""" + + @pytest.mark.asyncio + async def test_implementation_started_log_emitted_with_extra_fields(self, caplog): + """Structured log is emitted with correct extra fields.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add retry 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-99", + current_task_key="TASK-100", + tasks_by_repo={"acme/backend": ["TASK-100"]}, + ) + ) + + # Find the implementation_started log record + started_records = [ + r for r in caplog.records if "Implementation started for task" in r.message + ] + assert len(started_records) == 1, "Expected exactly one implementation_started log" + + record = started_records[0] + assert record.levelname == "INFO" + assert "TASK-100" in record.message + assert record.event == "implementation_started" + assert record.task_name == "Add retry logic" + assert record.feature_id == "FEAT-99" + assert record.task_id == "TASK-100" + + @pytest.mark.asyncio + async def test_implementation_started_log_empty_summary(self, caplog): + """Empty task summary results in task_name="" in extra dict.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="", description="Some description") + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="BUG-50", + current_task_key="TASK-200", + tasks_by_repo={"acme/backend": ["TASK-200"]}, + ) + ) + + started_records = [ + r for r in caplog.records if "Implementation started for task" in r.message + ] + assert len(started_records) == 1 + + record = started_records[0] + assert record.task_name == "" + + @pytest.mark.asyncio + async def test_implementation_started_log_special_characters_preserved(self, caplog): + """Special characters in task summary are logged as-is.""" + from forge.workflow.nodes.implementation import implement_task + + special_summary = "Fix: & unicode: 日本語" + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-1", + current_task_key="TASK-1", + tasks_by_repo={"acme/backend": ["TASK-1"]}, + ) + ) + + started_records = [ + r for r in caplog.records if "Implementation started for task" in r.message + ] + assert len(started_records) == 1 + + record = started_records[0] + assert record.task_name == special_summary From 3e23d0ecd86fede265d97217d1f0236cbd7afac4 Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 16:53:49 +0000 Subject: [PATCH 3/8] [AISOS-2202] Add structured end log on successful implementation Add INFO-level structured log message when implementation completes successfully in implement_task function: - Log message format: 'Implementation completed for task {current_task}' - Includes structured extra fields: event='implementation_completed', task_name, feature_id, task_id, success=True - Placed after container success log and before push/status comment for accurate timing Closes: AISOS-2202 --- src/forge/workflow/nodes/implementation.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py index 1fd7c0b4..39d139eb 100644 --- a/src/forge/workflow/nodes/implementation.py +++ b/src/forge/workflow/nodes/implementation.py @@ -223,6 +223,17 @@ 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, + }, + ) + # Persist each task commit before checkpointing. A subsequent task # or local review may resume on a worker with a different filesystem. try: From fa5d139bef2674542d3a76a4b5b5d3a8c7bf03f0 Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 16:56:50 +0000 Subject: [PATCH 4/8] [AISOS-2202] Add tests for implementation_completed structured log Add TestImplementationCompletedStructuredLog test class to verify the structured log emitted when implementation completes successfully: - test_implementation_completed_log_emitted_with_extra_fields: Verifies log message and all extra fields (event, task_name, feature_id, task_id, success=True) - test_implementation_completed_log_empty_summary: Verifies empty summary logs task_name="" - test_implementation_completed_log_not_emitted_on_failure: Verifies log is NOT emitted when container fails Addresses reviewer feedback requiring tests for the implementation_completed log added as part of the structured logging feature. Closes: AISOS-2202 --- .../workflow/nodes/test_implementation.py | 124 ++++++++++++++++++ 1 file changed, 124 insertions(+) diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index cdf71377..4c52ef4d 100644 --- a/tests/unit/workflow/nodes/test_implementation.py +++ b/tests/unit/workflow/nodes/test_implementation.py @@ -489,3 +489,127 @@ async def test_implementation_started_log_special_characters_preserved(self, cap record = started_records[0] assert record.task_name == special_summary + + +class TestImplementationCompletedStructuredLog: + """Tests for the structured log emitted when implementation completes successfully.""" + + @pytest.mark.asyncio + async def test_implementation_completed_log_emitted_with_extra_fields(self, caplog): + """Structured log is emitted with correct extra fields on success.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add retry 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-99", + current_task_key="TASK-100", + tasks_by_repo={"acme/backend": ["TASK-100"]}, + ) + ) + + # Find the implementation_completed log record + completed_records = [ + r for r in caplog.records if "Implementation completed for task" in r.message + ] + assert len(completed_records) == 1, "Expected exactly one implementation_completed log" + + record = completed_records[0] + assert record.levelname == "INFO" + assert "TASK-100" in record.message + assert record.event == "implementation_completed" + assert record.task_name == "Add retry logic" + assert record.feature_id == "FEAT-99" + assert record.task_id == "TASK-100" + assert record.success is True + + @pytest.mark.asyncio + async def test_implementation_completed_log_empty_summary(self, caplog): + """Empty task summary results in task_name="" in extra dict.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="", description="Some description") + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="BUG-50", + current_task_key="TASK-200", + tasks_by_repo={"acme/backend": ["TASK-200"]}, + ) + ) + + completed_records = [ + r for r in caplog.records if "Implementation completed for task" in r.message + ] + assert len(completed_records) == 1 + + record = completed_records[0] + assert record.task_name == "" + assert record.success is True + + @pytest.mark.asyncio + async def test_implementation_completed_log_not_emitted_on_failure(self, caplog): + """The implementation_completed log is not emitted when container fails.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add retry logic") + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-99", + current_task_key="TASK-100", + tasks_by_repo={"acme/backend": ["TASK-100"]}, + ) + ) + + # The implementation_completed log should NOT be emitted + completed_records = [ + r for r in caplog.records if "Implementation completed for task" in r.message + ] + assert len(completed_records) == 0, ( + "implementation_completed log should not be emitted on failure" + ) From f5a6bb0fb8bb271adb36c0e90fd75ca900ea919c Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 16:59:03 +0000 Subject: [PATCH 5/8] [AISOS-2203] Add structured end log on failed implementation Added structured INFO-level logging when implementation fails: - Initialize task_summary = "unknown" before Jira fetch to handle early failures - Added implementation_ended log in exception handler with success=False - Log includes: event, task_name, feature_id, task_id, success fields - Handle case where current_task is None using "unknown" placeholder - Log emitted before return statement to ensure timing accuracy Closes: AISOS-2203 --- src/forge/workflow/nodes/implementation.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py index 39d139eb..bd5a8299 100644 --- a/src/forge/workflow/nodes/implementation.py +++ b/src/forge/workflow/nodes/implementation.py @@ -166,6 +166,9 @@ async def implement_task(state: WorkflowState) -> WorkflowState: settings = get_settings() jira = JiraClient(settings) + # Initialize task_summary before Jira fetch to handle early failures (BR-003) + task_summary = "unknown" + try: # Get Task details from Jira task_issue = await jira.get_issue(current_task) @@ -285,6 +288,17 @@ async def implement_task(state: WorkflowState) -> WorkflowState: except Exception as e: logger.error(f"Implementation failed for {current_task}: {e}") + # Log end event before error handling for timing accuracy + 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), From c3ee784d421b5aec68dfa35ff6e9f082d177a4d9 Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 17:02:09 +0000 Subject: [PATCH 6/8] [AISOS-2203] Add tests for implementation_ended structured log on failure Add TestImplementationEndedStructuredLog class with four tests that verify: - implementation_ended log is emitted with correct extra fields on failure - task_name uses "unknown" placeholder when failure occurs before Jira fetch - task_id uses the selected task when available, even on failure - implementation_ended log is NOT emitted on success This addresses reviewer feedback requiring tests for the structured end log behavior added for failed implementations. Closes: AISOS-2203 --- .../workflow/nodes/test_implementation.py | 165 ++++++++++++++++++ 1 file changed, 165 insertions(+) diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index 4c52ef4d..897c8191 100644 --- a/tests/unit/workflow/nodes/test_implementation.py +++ b/tests/unit/workflow/nodes/test_implementation.py @@ -613,3 +613,168 @@ async def test_implementation_completed_log_not_emitted_on_failure(self, caplog) assert len(completed_records) == 0, ( "implementation_completed log should not be emitted on failure" ) + + +class TestImplementationEndedStructuredLog: + """Tests for the structured log emitted when implementation fails.""" + + @pytest.mark.asyncio + async def test_implementation_ended_log_emitted_on_failure(self, caplog): + """Structured log is emitted with correct extra fields on failure.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add retry logic") + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-99", + current_task_key="TASK-100", + tasks_by_repo={"acme/backend": ["TASK-100"]}, + ) + ) + + # Find the implementation_ended log record + ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] + assert len(ended_records) == 1, "Expected exactly one implementation_ended log" + + record = ended_records[0] + assert record.levelname == "INFO" + assert "TASK-100" in record.message + assert record.event == "implementation_ended" + assert record.task_name == "Add retry logic" + assert record.feature_id == "FEAT-99" + assert record.task_id == "TASK-100" + assert record.success is False + + @pytest.mark.asyncio + async def test_implementation_ended_log_with_unknown_task_summary_on_early_failure( + self, caplog + ): + """Uses 'unknown' for task_name when failure occurs before Jira fetch.""" + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + result = await implement_task( + _make_state( + ticket_key="FEAT-99", + current_task_key="TASK-100", + tasks_by_repo={"acme/backend": ["TASK-100"]}, + ) + ) + + # Verify failure was recorded + assert "Jira connection failed" in result["last_error"] + + # Find the implementation_ended log record + ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] + assert len(ended_records) == 1, "Expected exactly one implementation_ended log" + + record = ended_records[0] + assert record.levelname == "INFO" + assert record.event == "implementation_ended" + assert record.task_name == "unknown" # Initialized placeholder before Jira fetch + assert record.feature_id == "FEAT-99" + assert record.task_id == "TASK-100" + assert record.success is False + + @pytest.mark.asyncio + async def test_implementation_ended_log_with_unknown_task_id_placeholder(self, caplog): + """Uses 'unknown' for task_id when current_task is None.""" + from forge.workflow.nodes.implementation import implement_task + + # Create state where current_task is not set initially and task selection fails + mock_jira = AsyncMock() + mock_jira.get_issue = AsyncMock(side_effect=Exception("Task fetch failed")) + mock_jira.close = AsyncMock() + + # State where current_task needs to be selected, but then fails at Jira fetch + state = _make_state( + ticket_key="FEAT-99", + current_task_key=None, + tasks_by_repo={"acme/backend": ["TASK-100"]}, + ) + state["task_keys"] = ["TASK-100"] + + with ( + patch( + "forge.workflow.nodes.implementation.JiraClient", + return_value=mock_jira, + ), + patch("forge.workflow.nodes.implementation.get_settings"), + caplog.at_level("INFO", logger="forge.workflow.nodes.implementation"), + ): + result = await implement_task(state) + + # Verify failure was recorded + assert "Task fetch failed" in result["last_error"] + + # Find the implementation_ended log record + ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] + assert len(ended_records) == 1, "Expected exactly one implementation_ended log" + + record = ended_records[0] + assert record.event == "implementation_ended" + # task_id should be the selected task (TASK-100), not "unknown" + # The task selection happens before the try block + assert record.task_id == "TASK-100" + assert record.success is False + + @pytest.mark.asyncio + async def test_implementation_ended_log_not_emitted_on_success(self, caplog): + """The implementation_ended log is NOT emitted when implementation succeeds.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add retry 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-99", + current_task_key="TASK-100", + tasks_by_repo={"acme/backend": ["TASK-100"]}, + ) + ) + + # The implementation_ended log should NOT be emitted on success + ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] + assert len(ended_records) == 0, "implementation_ended log should not be emitted on success" From ba2e1510b2a9b3098fac0fb8eda5b1a286cc2e87 Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 17:06:52 +0000 Subject: [PATCH 7/8] [AISOS-2204] Add unit tests for structured logging in implement_task Added TestImplementTaskStructuredLogging test class with 7 test methods: - test_logs_implementation_started_with_structured_fields (TS-001) - test_logs_implementation_completed_on_success (TS-002) - test_logs_implementation_ended_on_failure (TS-003) - test_logs_implementation_ended_on_exception (TS-004) - test_logs_unknown_task_name_when_jira_fails_early (TS-005) - test_logs_empty_task_summary_as_empty_string (TS-006) - test_logs_special_characters_in_task_summary (TS-007) Tests verify: - Log level is INFO - Presence of event, task_name, feature_id, task_id fields - success field on end logs (True for completed, False for ended) - Placeholder handling when Jira fails early - Empty summary handling (logs empty string) - Special characters are not escaped All 29 tests pass. Tests follow existing patterns using _make_state(), _make_mock_jira(), and caplog fixture. Closes: AISOS-2204 --- .../workflow/nodes/test_implementation.py | 290 ++++++++++++++++++ 1 file changed, 290 insertions(+) diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index 897c8191..2b245ad5 100644 --- a/tests/unit/workflow/nodes/test_implementation.py +++ b/tests/unit/workflow/nodes/test_implementation.py @@ -778,3 +778,293 @@ async def test_implementation_ended_log_not_emitted_on_success(self, caplog): # The implementation_ended log should NOT be emitted on success ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] assert len(ended_records) == 0, "implementation_ended log should not be emitted on success" + + +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: Verify start log has correct extra fields.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Fix authentication bug") + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-123", + current_task_key="TASK-456", + tasks_by_repo={"acme/backend": ["TASK-456"]}, + ) + ) + + started_records = [ + r for r in caplog.records if "Implementation started for task" in r.message + ] + assert len(started_records) == 1 + + record = started_records[0] + assert record.levelname == "INFO" + assert record.event == "implementation_started" + assert record.task_name == "Fix authentication bug" + assert record.feature_id == "FEAT-123" + assert record.task_id == "TASK-456" + + @pytest.mark.asyncio + async def test_logs_implementation_completed_on_success(self, caplog): + """TS-002: Verify success end log has success=True.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add caching layer") + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-200", + current_task_key="TASK-300", + tasks_by_repo={"acme/backend": ["TASK-300"]}, + ) + ) + + completed_records = [ + r for r in caplog.records if "Implementation completed for task" in r.message + ] + assert len(completed_records) == 1 + + record = completed_records[0] + assert record.levelname == "INFO" + assert record.event == "implementation_completed" + assert record.task_name == "Add caching layer" + assert record.feature_id == "FEAT-200" + assert record.task_id == "TASK-300" + assert record.success is True + + @pytest.mark.asyncio + async def test_logs_implementation_ended_on_failure(self, caplog): + """TS-003: Verify failure end log has success=False.""" + 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 = "container crashed" + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-400", + current_task_key="TASK-500", + tasks_by_repo={"acme/backend": ["TASK-500"]}, + ) + ) + + ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] + assert len(ended_records) == 1 + + record = ended_records[0] + assert record.levelname == "INFO" + assert record.event == "implementation_ended" + assert record.task_name == "Refactor database layer" + assert record.feature_id == "FEAT-400" + assert record.task_id == "TASK-500" + assert record.success is False + + @pytest.mark.asyncio + async def test_logs_implementation_ended_on_exception(self, caplog): + """TS-004: Verify exception path emits end log.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Update API endpoints") + runner = MagicMock() + runner.run = AsyncMock(side_effect=RuntimeError("unexpected error")) + + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + result = await implement_task( + _make_state( + ticket_key="FEAT-600", + current_task_key="TASK-700", + tasks_by_repo={"acme/backend": ["TASK-700"]}, + ) + ) + + # Verify failure was captured in state + assert "unexpected error" in result["last_error"] + + ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] + assert len(ended_records) == 1 + + record = ended_records[0] + assert record.levelname == "INFO" + assert record.event == "implementation_ended" + assert record.task_name == "Update API endpoints" + assert record.feature_id == "FEAT-600" + assert record.task_id == "TASK-700" + assert record.success is False + + @pytest.mark.asyncio + async def test_logs_unknown_task_name_when_jira_fails_early(self, caplog): + """TS-005: Verify placeholder handling when Jira fetch fails early.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = AsyncMock() + mock_jira.get_issue = AsyncMock(side_effect=Exception("Jira unavailable")) + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + result = await implement_task( + _make_state( + ticket_key="FEAT-800", + current_task_key="TASK-900", + tasks_by_repo={"acme/backend": ["TASK-900"]}, + ) + ) + + # Verify failure was recorded + assert "Jira unavailable" in result["last_error"] + + ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] + assert len(ended_records) == 1 + + record = ended_records[0] + assert record.levelname == "INFO" + assert record.event == "implementation_ended" + assert record.task_name == "unknown" + assert record.feature_id == "FEAT-800" + assert record.task_id == "TASK-900" + assert record.success is False + + @pytest.mark.asyncio + async def test_logs_empty_task_summary_as_empty_string(self, caplog): + """TS-006: Verify empty summary handling - logs empty string.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="", description="Some task description") + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-1000", + current_task_key="TASK-1100", + tasks_by_repo={"acme/backend": ["TASK-1100"]}, + ) + ) + + # Check started log + started_records = [ + r for r in caplog.records if "Implementation started for task" in r.message + ] + assert len(started_records) == 1 + assert started_records[0].task_name == "" + + # Check completed log + completed_records = [ + r for r in caplog.records if "Implementation completed for task" in r.message + ] + assert len(completed_records) == 1 + assert completed_records[0].task_name == "" + + @pytest.mark.asyncio + async def test_logs_special_characters_in_task_summary(self, caplog): + """TS-007: Verify special chars not escaped in task_name field.""" + from forge.workflow.nodes.implementation import implement_task + + special_summary = "Fix: & unicode: 日本語 © ® ™" + 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("INFO", logger="forge.workflow.nodes.implementation"), + ): + await implement_task( + _make_state( + ticket_key="FEAT-1200", + current_task_key="TASK-1300", + tasks_by_repo={"acme/backend": ["TASK-1300"]}, + ) + ) + + started_records = [ + r for r in caplog.records if "Implementation started for task" in r.message + ] + assert len(started_records) == 1 + + record = started_records[0] + # Special characters should be preserved as-is, not escaped + assert record.task_name == special_summary From dc36fc6bda384101c2bc61c05ff20fd148beefc0 Mon Sep 17 00:00:00 2001 From: Forge Date: Thu, 16 Jul 2026 17:09:48 +0000 Subject: [PATCH 8/8] [AISOS-2204] Remove redundant test classes for structured logging Removed 3 redundant test classes per reviewer feedback: - TestImplementationStartedStructuredLog - TestImplementationCompletedStructuredLog - TestImplementationEndedStructuredLog These classes duplicated tests already present in the required TestImplementTaskStructuredLogging class. The consolidated class contains all 7 test methods per specification TS-001 through TS-007. Net reduction of ~377 lines of redundant test code. Closes: AISOS-2204 --- .../workflow/nodes/test_implementation.py | 408 ------------------ 1 file changed, 408 deletions(-) diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index 2b245ad5..ed1b8c9d 100644 --- a/tests/unit/workflow/nodes/test_implementation.py +++ b/tests/unit/workflow/nodes/test_implementation.py @@ -372,414 +372,6 @@ async def test_bug_container_failure_keeps_bug_implementation_node(self): assert result["retry_count"] == 1 -class TestImplementationStartedStructuredLog: - """Tests for the structured log emitted when implementation starts.""" - - @pytest.mark.asyncio - async def test_implementation_started_log_emitted_with_extra_fields(self, caplog): - """Structured log is emitted with correct extra fields.""" - from forge.workflow.nodes.implementation import implement_task - - mock_jira = _make_mock_jira(summary="Add retry 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="FEAT-99", - current_task_key="TASK-100", - tasks_by_repo={"acme/backend": ["TASK-100"]}, - ) - ) - - # Find the implementation_started log record - started_records = [ - r for r in caplog.records if "Implementation started for task" in r.message - ] - assert len(started_records) == 1, "Expected exactly one implementation_started log" - - record = started_records[0] - assert record.levelname == "INFO" - assert "TASK-100" in record.message - assert record.event == "implementation_started" - assert record.task_name == "Add retry logic" - assert record.feature_id == "FEAT-99" - assert record.task_id == "TASK-100" - - @pytest.mark.asyncio - async def test_implementation_started_log_empty_summary(self, caplog): - """Empty task summary results in task_name="" in extra dict.""" - from forge.workflow.nodes.implementation import implement_task - - mock_jira = _make_mock_jira(summary="", description="Some description") - 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="BUG-50", - current_task_key="TASK-200", - tasks_by_repo={"acme/backend": ["TASK-200"]}, - ) - ) - - started_records = [ - r for r in caplog.records if "Implementation started for task" in r.message - ] - assert len(started_records) == 1 - - record = started_records[0] - assert record.task_name == "" - - @pytest.mark.asyncio - async def test_implementation_started_log_special_characters_preserved(self, caplog): - """Special characters in task summary are logged as-is.""" - from forge.workflow.nodes.implementation import implement_task - - special_summary = "Fix: & unicode: 日本語" - 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="FEAT-1", - current_task_key="TASK-1", - tasks_by_repo={"acme/backend": ["TASK-1"]}, - ) - ) - - started_records = [ - r for r in caplog.records if "Implementation started for task" in r.message - ] - assert len(started_records) == 1 - - record = started_records[0] - assert record.task_name == special_summary - - -class TestImplementationCompletedStructuredLog: - """Tests for the structured log emitted when implementation completes successfully.""" - - @pytest.mark.asyncio - async def test_implementation_completed_log_emitted_with_extra_fields(self, caplog): - """Structured log is emitted with correct extra fields on success.""" - from forge.workflow.nodes.implementation import implement_task - - mock_jira = _make_mock_jira(summary="Add retry 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="FEAT-99", - current_task_key="TASK-100", - tasks_by_repo={"acme/backend": ["TASK-100"]}, - ) - ) - - # Find the implementation_completed log record - completed_records = [ - r for r in caplog.records if "Implementation completed for task" in r.message - ] - assert len(completed_records) == 1, "Expected exactly one implementation_completed log" - - record = completed_records[0] - assert record.levelname == "INFO" - assert "TASK-100" in record.message - assert record.event == "implementation_completed" - assert record.task_name == "Add retry logic" - assert record.feature_id == "FEAT-99" - assert record.task_id == "TASK-100" - assert record.success is True - - @pytest.mark.asyncio - async def test_implementation_completed_log_empty_summary(self, caplog): - """Empty task summary results in task_name="" in extra dict.""" - from forge.workflow.nodes.implementation import implement_task - - mock_jira = _make_mock_jira(summary="", description="Some description") - 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="BUG-50", - current_task_key="TASK-200", - tasks_by_repo={"acme/backend": ["TASK-200"]}, - ) - ) - - completed_records = [ - r for r in caplog.records if "Implementation completed for task" in r.message - ] - assert len(completed_records) == 1 - - record = completed_records[0] - assert record.task_name == "" - assert record.success is True - - @pytest.mark.asyncio - async def test_implementation_completed_log_not_emitted_on_failure(self, caplog): - """The implementation_completed log is not emitted when container fails.""" - from forge.workflow.nodes.implementation import implement_task - - mock_jira = _make_mock_jira(summary="Add retry logic") - 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="FEAT-99", - current_task_key="TASK-100", - tasks_by_repo={"acme/backend": ["TASK-100"]}, - ) - ) - - # The implementation_completed log should NOT be emitted - completed_records = [ - r for r in caplog.records if "Implementation completed for task" in r.message - ] - assert len(completed_records) == 0, ( - "implementation_completed log should not be emitted on failure" - ) - - -class TestImplementationEndedStructuredLog: - """Tests for the structured log emitted when implementation fails.""" - - @pytest.mark.asyncio - async def test_implementation_ended_log_emitted_on_failure(self, caplog): - """Structured log is emitted with correct extra fields on failure.""" - from forge.workflow.nodes.implementation import implement_task - - mock_jira = _make_mock_jira(summary="Add retry logic") - 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="FEAT-99", - current_task_key="TASK-100", - tasks_by_repo={"acme/backend": ["TASK-100"]}, - ) - ) - - # Find the implementation_ended log record - ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] - assert len(ended_records) == 1, "Expected exactly one implementation_ended log" - - record = ended_records[0] - assert record.levelname == "INFO" - assert "TASK-100" in record.message - assert record.event == "implementation_ended" - assert record.task_name == "Add retry logic" - assert record.feature_id == "FEAT-99" - assert record.task_id == "TASK-100" - assert record.success is False - - @pytest.mark.asyncio - async def test_implementation_ended_log_with_unknown_task_summary_on_early_failure( - self, caplog - ): - """Uses 'unknown' for task_name when failure occurs before Jira fetch.""" - 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - result = await implement_task( - _make_state( - ticket_key="FEAT-99", - current_task_key="TASK-100", - tasks_by_repo={"acme/backend": ["TASK-100"]}, - ) - ) - - # Verify failure was recorded - assert "Jira connection failed" in result["last_error"] - - # Find the implementation_ended log record - ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] - assert len(ended_records) == 1, "Expected exactly one implementation_ended log" - - record = ended_records[0] - assert record.levelname == "INFO" - assert record.event == "implementation_ended" - assert record.task_name == "unknown" # Initialized placeholder before Jira fetch - assert record.feature_id == "FEAT-99" - assert record.task_id == "TASK-100" - assert record.success is False - - @pytest.mark.asyncio - async def test_implementation_ended_log_with_unknown_task_id_placeholder(self, caplog): - """Uses 'unknown' for task_id when current_task is None.""" - from forge.workflow.nodes.implementation import implement_task - - # Create state where current_task is not set initially and task selection fails - mock_jira = AsyncMock() - mock_jira.get_issue = AsyncMock(side_effect=Exception("Task fetch failed")) - mock_jira.close = AsyncMock() - - # State where current_task needs to be selected, but then fails at Jira fetch - state = _make_state( - ticket_key="FEAT-99", - current_task_key=None, - tasks_by_repo={"acme/backend": ["TASK-100"]}, - ) - state["task_keys"] = ["TASK-100"] - - with ( - patch( - "forge.workflow.nodes.implementation.JiraClient", - return_value=mock_jira, - ), - patch("forge.workflow.nodes.implementation.get_settings"), - caplog.at_level("INFO", logger="forge.workflow.nodes.implementation"), - ): - result = await implement_task(state) - - # Verify failure was recorded - assert "Task fetch failed" in result["last_error"] - - # Find the implementation_ended log record - ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] - assert len(ended_records) == 1, "Expected exactly one implementation_ended log" - - record = ended_records[0] - assert record.event == "implementation_ended" - # task_id should be the selected task (TASK-100), not "unknown" - # The task selection happens before the try block - assert record.task_id == "TASK-100" - assert record.success is False - - @pytest.mark.asyncio - async def test_implementation_ended_log_not_emitted_on_success(self, caplog): - """The implementation_ended log is NOT emitted when implementation succeeds.""" - from forge.workflow.nodes.implementation import implement_task - - mock_jira = _make_mock_jira(summary="Add retry 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("INFO", logger="forge.workflow.nodes.implementation"), - ): - await implement_task( - _make_state( - ticket_key="FEAT-99", - current_task_key="TASK-100", - tasks_by_repo={"acme/backend": ["TASK-100"]}, - ) - ) - - # The implementation_ended log should NOT be emitted on success - ended_records = [r for r in caplog.records if "Implementation ended for task" in r.message] - assert len(ended_records) == 0, "implementation_ended log should not be emitted on success" - - class TestImplementTaskStructuredLogging: """Consolidated tests for structured logging in implement_task (TS-001 through TS-007)."""