From 25a7bedf645df8ec5f41b8d564501b16aa06dc52 Mon Sep 17 00:00:00 2001 From: Forge Date: Tue, 21 Jul 2026 06:56:17 +0000 Subject: [PATCH 1/3] [AISOS-2257] Add structured logging at task implementation start Add INFO-level log emission when task implementation begins: - Initialize feature_id, task_id_log, task_name with "N/A" defaults - Update variables with actual values after Jira task retrieval - Emit log with format: Implementation started: task_name="{name}" feature_id="{id}" task_id="{id}" - Log is emitted before container execution begins Also add unit tests for the new logging behavior: - test_logs_implementation_started_with_correct_format - test_start_log_emitted_before_container_execution - test_start_log_uses_na_defaults_when_values_missing Closes: AISOS-2257 --- src/forge/workflow/nodes/implementation.py | 17 +++ .../workflow/nodes/test_implementation.py | 119 ++++++++++++++++++ 2 files changed, 136 insertions(+) diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py index 5514cc26..7c3932b5 100644 --- a/src/forge/workflow/nodes/implementation.py +++ b/src/forge/workflow/nodes/implementation.py @@ -166,12 +166,29 @@ async def implement_task(state: WorkflowState) -> WorkflowState: settings = get_settings() jira = JiraClient(settings) + # Initialize logging variables with defaults + feature_id = "N/A" + task_id_log = "N/A" + task_name = "N/A" + 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 + # Update logging variables with actual values + feature_id = state.get("ticket_key") or "N/A" + task_id_log = current_task or "N/A" + task_name = task_summary or "N/A" + + logger.info( + 'Implementation started: task_name="%s" feature_id="%s" task_id="%s"', + task_name, + feature_id, + task_id_log, + ) + # Post status comment at task implementation start await post_status_comment( jira, diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index 90de8ba8..04d0dc93 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 TestImplementationStartLogging: + """Tests for structured logging at implementation start.""" + + @pytest.mark.asyncio + async def test_logs_implementation_started_with_correct_format(self, caplog): + """Implementation started log is emitted with correct values.""" + 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("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"]}, + ) + ) + + # Verify the start log was emitted with correct format + start_logs = [r for r in caplog.records if "Implementation started:" in r.message] + assert len(start_logs) == 1 + log_record = start_logs[0] + assert 'task_name="Fix null pointer in AuthService"' in log_record.message + assert 'feature_id="FEAT-99"' in log_record.message + assert 'task_id="TASK-100"' in log_record.message + + @pytest.mark.asyncio + async def test_start_log_emitted_before_container_execution(self, caplog): + """Start log is emitted before container runner is called.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Add retry logic") + runner = _make_successful_runner() + run_call_time = None + + async def track_run_call(*_args, **_kwargs): + nonlocal run_call_time + run_call_time = len(caplog.records) + result = MagicMock() + result.success = True + result.error_message = None + return result + + runner.run = track_run_call + + 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()) + + # Find the start log position + start_log_positions = [ + i for i, r in enumerate(caplog.records) if "Implementation started:" in r.message + ] + assert len(start_log_positions) == 1 + start_log_position = start_log_positions[0] + + # Verify start log was emitted before container run was called + assert start_log_position < run_call_time + + @pytest.mark.asyncio + async def test_start_log_uses_na_defaults_when_values_missing(self, caplog): + """Start log falls back to N/A for missing values (edge case test).""" + from forge.workflow.nodes.implementation import implement_task + + # Create a mock that returns None/empty for summary + mock_jira = AsyncMock() + issue = MagicMock() + issue.summary = None + issue.description = None + mock_jira.get_issue = AsyncMock(return_value=issue) + mock_jira.add_comment = AsyncMock() + mock_jira.close = AsyncMock() + + 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()) + + # Verify the start log uses N/A for None task_name + start_logs = [r for r in caplog.records if "Implementation started:" in r.message] + assert len(start_logs) == 1 + log_record = start_logs[0] + assert 'task_name="N/A"' in log_record.message From cfb632e0ee6a1dd2a7194ba042c678c231f1ebf1 Mon Sep 17 00:00:00 2001 From: Forge Date: Tue, 21 Jul 2026 07:04:10 +0000 Subject: [PATCH 2/3] [AISOS-2258] Add end log to implementation try/finally block Added structured logging at the end of task implementation using the existing finally block. The end log is emitted at INFO level with the format: Implementation completed: task_name="{task_name}" feature_id="{feature_id}" task_id="{task_id}" Changes: - Added logger.info() call in the finally block before jira.close() - Uses the same logging variables (task_name, feature_id, task_id_log) initialized with "N/A" defaults from AISOS-2257 The finally block ensures the end log is emitted on both success and failure paths, providing symmetric bookends with the start log. Closes: AISOS-2258 --- src/forge/workflow/nodes/implementation.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/src/forge/workflow/nodes/implementation.py b/src/forge/workflow/nodes/implementation.py index 7c3932b5..9381be6e 100644 --- a/src/forge/workflow/nodes/implementation.py +++ b/src/forge/workflow/nodes/implementation.py @@ -288,6 +288,12 @@ async def implement_task(state: WorkflowState) -> WorkflowState: "retry_count": state.get("retry_count", 0) + 1, } finally: + logger.info( + 'Implementation completed: task_name="%s" feature_id="%s" task_id="%s"', + task_name, + feature_id, + task_id_log, + ) await jira.close() From 6d367fe1993aed7fca48eb016be9782265391cbb Mon Sep 17 00:00:00 2001 From: Forge Date: Tue, 21 Jul 2026 07:08:16 +0000 Subject: [PATCH 3/3] [AISOS-2259] Add unit tests for structured logging in implement_task Added TestImplementationStructuredLogging test class with 6 test methods covering all structured logging scenarios for the implement_task workflow node: 1. test_start_log_emitted_with_task_details - Verifies INFO log with 'Implementation started' contains correct task_name, feature_id, task_id 2. test_end_log_emitted_on_success - Verifies INFO log with 'Implementation completed' after successful container run 3. test_end_log_emitted_on_failure - Mocks container to return success=False, verifies end log still emitted via finally block 4. test_feature_id_fallback_to_na - Creates state with ticket_key=None, verifies feature_id='N/A' in logs 5. test_task_name_fallback_to_na - Mocks Jira to raise exception before summary retrieval, verifies task_name='N/A' in end log 6. test_no_logs_when_no_current_task - State with current_task_key=None and empty task_keys, verifies no 'Implementation started/completed' logs (early exit path) All tests use: - Existing test fixtures (_make_state, _make_mock_jira, _make_successful_runner) - caplog fixture for log capture at INFO level - Existing mock patterns with patch() for JiraClient, ContainerRunner, get_settings Closes: AISOS-2259 --- .../workflow/nodes/test_implementation.py | 241 ++++++++++++++++++ 1 file changed, 241 insertions(+) diff --git a/tests/unit/workflow/nodes/test_implementation.py b/tests/unit/workflow/nodes/test_implementation.py index 04d0dc93..a35a3c8d 100644 --- a/tests/unit/workflow/nodes/test_implementation.py +++ b/tests/unit/workflow/nodes/test_implementation.py @@ -489,3 +489,244 @@ async def test_start_log_uses_na_defaults_when_values_missing(self, caplog): assert len(start_logs) == 1 log_record = start_logs[0] assert 'task_name="N/A"' in log_record.message + + +class TestImplementationStructuredLogging: + """Tests for structured logging at implementation start and end.""" + + @pytest.mark.asyncio + async def test_start_log_emitted_with_task_details(self, caplog): + """Verify INFO log with 'Implementation started' contains correct task_name, feature_id, task_id.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Implement user authentication") + 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-201", + tasks_by_repo={"acme/backend": ["TASK-201"]}, + ) + ) + + # Verify the start log was emitted with correct values + start_logs = [r for r in caplog.records if "Implementation started" in r.message] + assert len(start_logs) == 1 + log_record = start_logs[0] + assert log_record.levelname == "INFO" + assert 'task_name="Implement user authentication"' in log_record.message + assert 'feature_id="FEAT-200"' in log_record.message + assert 'task_id="TASK-201"' in log_record.message + + @pytest.mark.asyncio + async def test_end_log_emitted_on_success(self, caplog): + """Verify INFO log with 'Implementation completed' after successful container run.""" + 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"), + ): + result = await implement_task( + _make_state( + ticket_key="FEAT-300", + current_task_key="TASK-301", + tasks_by_repo={"acme/backend": ["TASK-301"]}, + ) + ) + + # Verify success + assert result["last_error"] is None + + # Verify the end log was emitted + end_logs = [r for r in caplog.records if "Implementation completed" in r.message] + assert len(end_logs) == 1 + log_record = end_logs[0] + assert log_record.levelname == "INFO" + assert 'task_name="Add caching layer"' in log_record.message + assert 'feature_id="FEAT-300"' in log_record.message + assert 'task_id="TASK-301"' in log_record.message + + @pytest.mark.asyncio + async def test_end_log_emitted_on_failure(self, caplog): + """Mock container to return success=False, verify end log still emitted.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Failing task") + runner = MagicMock() + container_result = MagicMock() + container_result.success = False + container_result.error_message = "container execution 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"), + ): + result = await implement_task( + _make_state( + ticket_key="FEAT-400", + current_task_key="TASK-401", + tasks_by_repo={"acme/backend": ["TASK-401"]}, + ) + ) + + # Verify failure + assert result["last_error"] == "container execution failed" + + # Verify the end log was still emitted (in finally block) + end_logs = [r for r in caplog.records if "Implementation completed" in r.message] + assert len(end_logs) == 1 + log_record = end_logs[0] + assert log_record.levelname == "INFO" + assert 'task_name="Failing task"' in log_record.message + assert 'feature_id="FEAT-400"' in log_record.message + assert 'task_id="TASK-401"' in log_record.message + + @pytest.mark.asyncio + async def test_feature_id_fallback_to_na(self, caplog): + """Create state with ticket_key=None, verify feature_id='N/A' in logs.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = _make_mock_jira(summary="Task without feature") + runner = _make_successful_runner() + + state = _make_state( + ticket_key=None, + current_task_key="TASK-501", + tasks_by_repo={"acme/backend": ["TASK-501"]}, + ) + + 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(state) + + # Verify feature_id is N/A in start log + start_logs = [r for r in caplog.records if "Implementation started" in r.message] + assert len(start_logs) == 1 + assert 'feature_id="N/A"' in start_logs[0].message + + # Verify feature_id is N/A in end log + end_logs = [r for r in caplog.records if "Implementation completed" in r.message] + assert len(end_logs) == 1 + assert 'feature_id="N/A"' in end_logs[0].message + + @pytest.mark.asyncio + async def test_task_name_fallback_to_na(self, caplog): + """Mock Jira to raise exception before summary retrieval, verify task_name='N/A' in end log.""" + from forge.workflow.nodes.implementation import implement_task + + mock_jira = AsyncMock() + mock_jira.get_issue = AsyncMock(side_effect=Exception("Jira API unavailable")) + mock_jira.add_comment = AsyncMock() + mock_jira.close = AsyncMock() + + 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"), + ): + result = await implement_task( + _make_state( + ticket_key="FEAT-600", + current_task_key="TASK-601", + tasks_by_repo={"acme/backend": ["TASK-601"]}, + ) + ) + + # Verify error was captured + assert "Jira API unavailable" in result["last_error"] + + # Verify the end log was emitted with N/A values + # (because get_issue failed before logging variable assignments) + end_logs = [r for r in caplog.records if "Implementation completed" in r.message] + assert len(end_logs) == 1 + log_record = end_logs[0] + assert 'task_name="N/A"' in log_record.message + # All values remain at defaults since exception happened before assignments + assert 'feature_id="N/A"' in log_record.message + assert 'task_id="N/A"' in log_record.message + + @pytest.mark.asyncio + async def test_no_logs_when_no_current_task(self, caplog): + """State with current_task_key=None and empty task_keys, verify no 'Implementation started/completed' logs.""" + from forge.workflow.nodes.implementation import implement_task + + state = _make_state( + ticket_key="FEAT-700", + current_task_key=None, + ) + state["task_keys"] = [] + state["tasks_by_repo"] = {} + + mock_git = MagicMock() + mock_git.has_uncommitted_changes.return_value = False + + with ( + patch( + "forge.workflow.nodes.implementation.prepare_workspace", + return_value=(state["workspace_path"], mock_git), + ), + caplog.at_level("INFO", logger="forge.workflow.nodes.implementation"), + ): + result = await implement_task(state) + + # Verify the function took the early exit path + assert result["current_node"] == "local_review" + + # Verify no Implementation started/completed logs + start_logs = [r for r in caplog.records if "Implementation started" in r.message] + end_logs = [r for r in caplog.records if "Implementation completed" in r.message] + assert len(start_logs) == 0 + assert len(end_logs) == 0