diff --git a/.env.example b/.env.example index 4aecde790..dab6fdcae 100644 --- a/.env.example +++ b/.env.example @@ -255,6 +255,12 @@ LOG_LEVEL=INFO # LOG_FILE=/var/log/forge/forge.log # Maximum retry attempts for autonomous CI fixes CI_FIX_MAX_RETRIES=5 + +# Optional evidence-based report after successful or blocked terminal workflows. +RETROSPECTIVE_ENABLED=false +# Separate, explicit opt-in for deduplicated Jira improvement tasks. +RETROSPECTIVE_CREATE_ISSUES=false +RETROSPECTIVE_MAX_ITEMS=20 # Comma-separated substrings of CI check names to ignore (permanently-pending meta-checks). # Checks containing any of these substrings are skipped during CI evaluation. CI_IGNORED_CHECKS=tide diff --git a/docs/developer-guide.md b/docs/developer-guide.md index 9bf63a0e4..08b35a85d 100644 --- a/docs/developer-guide.md +++ b/docs/developer-guide.md @@ -197,6 +197,13 @@ CLICKHOUSE_PASSWORD=clickhouse # CI behaviour CI_FIX_MAX_RETRIES=5 + +# Terminal workflow retrospectives are opt-in. Reports contain only bounded, +# redacted facts and are published as Jira comments. Improvement task creation +# has a separate opt-in and uses stable labels to avoid duplicates. +RETROSPECTIVE_ENABLED=false +RETROSPECTIVE_CREATE_ISSUES=false +RETROSPECTIVE_MAX_ITEMS=20 CI_IGNORED_CHECKS=tide # comma-separated substrings of checks to ignore ``` diff --git a/src/forge/config.py b/src/forge/config.py index d109ce92e..4f952ed07 100644 --- a/src/forge/config.py +++ b/src/forge/config.py @@ -423,6 +423,23 @@ def model_policy_resolver(self): ci_fix_max_retries: int = Field( default=5, description="Maximum retry attempts for autonomous CI fixes" ) + retrospective_enabled: bool = Field( + default=False, + description="Publish a bounded evidence-based retrospective after terminal workflows", + ) + retrospective_create_issues: bool = Field( + default=False, + description=( + "Allow retrospective recommendations to create Jira improvement tasks. " + "Disabled independently of retrospective reporting by default." + ), + ) + retrospective_max_items: int = Field( + default=20, + ge=1, + le=100, + description="Maximum evidence items retained in a retrospective input", + ) ci_ignored_checks: str = Field( default="tide", description=( diff --git a/src/forge/orchestrator/worker.py b/src/forge/orchestrator/worker.py index b623bedbd..f80a42612 100644 --- a/src/forge/orchestrator/worker.py +++ b/src/forge/orchestrator/worker.py @@ -37,6 +37,7 @@ save_active_pull_request, ) from forge.workflow.registry import create_default_router +from forge.workflow.retrospective import run_retrospective from forge.workflow.router import WorkflowRouter from forge.workflow.utils.automated_review_triage import ( is_bot_sender, @@ -110,6 +111,17 @@ async def _report_new_workflow_error(result: dict, error_before_invoke: str | No } +async def _run_terminal_retrospective(state: dict[str, Any]) -> bool: + """Run optional analysis without allowing it to affect workflow completion.""" + try: + return await run_retrospective(state) is not None + except Exception: + logger.exception( + "Retrospective failed for %s; terminal outcome is unchanged", state.get("ticket_key") + ) + return False + + class OrchestratorWorker: """Worker that processes workflow events from Redis queue.""" @@ -422,6 +434,9 @@ async def _process_workflow(self, message: QueueMessage) -> None: f"'{updated_values.get('current_node')}', skipping invocation" ) await compiled_workflow.aupdate_state(config, updated_values) + if await _run_terminal_retrospective(updated_values): + updated_values["retrospective_completed"] = True + await compiled_workflow.aupdate_state(config, updated_values) return # If _handle_resume_event returned the state object unchanged (identity @@ -506,6 +521,11 @@ async def _process_workflow(self, message: QueueMessage) -> None: ticket_type = result.get("ticket_type", "unknown") record_workflow_completed(ticket_type=ticket_type, final_node=final_node) + is_terminal = final_node == "complete" or result.get("is_blocked", False) + if is_terminal and await _run_terminal_retrospective(result): + result["retrospective_completed"] = True + await compiled_workflow.aupdate_state(config, result) + except Exception as e: import traceback diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py index fcc4776c2..d58ae74b8 100644 --- a/src/forge/workflow/base.py +++ b/src/forge/workflow/base.py @@ -28,6 +28,7 @@ class BaseState(TypedDict, total=False): is_blocked: bool retry_count: int last_error: str | None + retrospective_completed: bool # Timestamps created_at: str diff --git a/src/forge/workflow/retrospective.py b/src/forge/workflow/retrospective.py new file mode 100644 index 000000000..db0ae7724 --- /dev/null +++ b/src/forge/workflow/retrospective.py @@ -0,0 +1,193 @@ +"""Bounded, non-mutating retrospective reporting for terminal workflows.""" + +from __future__ import annotations + +import hashlib +import logging +import re +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + +from forge.config import Settings, get_settings +from forge.integrations.jira.client import JiraClient + +logger = logging.getLogger(__name__) + +_SECRET = re.compile(r"(?i)(api[_-]?key|token|secret|password|authorization)\s*[:=]\s*[^\s,;]+") +_MAX_TEXT = 240 + + +def _safe_text(value: Any) -> str: + text = str("" if value is None else value)[:_MAX_TEXT] + return _SECRET.sub(r"\1=[REDACTED]", text) + + +@dataclass(frozen=True) +class Evidence: + stage: str + metric: str + value: str + + +@dataclass(frozen=True) +class RetrospectiveInput: + ticket_key: str + outcome: str + started_at: str | None + completed_at: str + evidence: tuple[Evidence, ...] + historical_runs: int = 0 + + +@dataclass(frozen=True) +class Recommendation: + title: str + detail: str + evidence: tuple[str, ...] + scope: str = "single incident" + + +@dataclass(frozen=True) +class RetrospectiveReport: + input: RetrospectiveInput + recommendations: tuple[Recommendation, ...] = field(default_factory=tuple) + model: str = "deterministic-v1" + input_tokens: int = 0 + output_tokens: int = 0 + estimated_cost_usd: float = 0.0 + + +def build_input(state: Mapping[str, Any], max_items: int = 20) -> RetrospectiveInput: + """Project unrestricted workflow state into a small, redacted typed summary.""" + evidence: list[Evidence] = [] + + def add(stage: str, metric: str, value: Any) -> None: + if value is not None and len(evidence) < max_items: + evidence.append(Evidence(stage, metric, _safe_text(value))) + + blocked = bool(state.get("is_blocked")) + outcome = "blocked" if blocked else "successful" + if state.get("last_error") and not blocked: + outcome = "partially_completed" + add("workflow", "terminal_node", state.get("current_node", "unknown")) + add("workflow", "retry_count", state.get("retry_count", 0)) + add("workflow", "blocked", blocked) + add("workflow", "error", state.get("last_error")) + add("ci", "status", state.get("ci_status")) + add("ci", "fix_attempts", state.get("ci_fix_attempt", 0)) + add("ci", "failed_checks", len(state.get("ci_failed_checks") or [])) + add("review", "local_attempts", state.get("local_review_attempts", 0)) + add("review", "ai_status", state.get("ai_review_status")) + add("review", "human_status", state.get("human_review_status")) + add("delivery", "repositories_completed", len(state.get("repos_completed") or [])) + add("delivery", "pull_requests", len(state.get("pr_urls") or [])) + return RetrospectiveInput( + ticket_key=_safe_text(state.get("ticket_key", "unknown")), + outcome=outcome, + started_at=state.get("created_at"), + completed_at=datetime.now(UTC).isoformat(), + evidence=tuple(evidence), + ) + + +def analyze(data: RetrospectiveInput) -> RetrospectiveReport: + """Create conservative recommendations directly supported by supplied facts.""" + values = {(item.stage, item.metric): item.value for item in data.evidence} + recs: list[Recommendation] = [] + scope = "recurring pattern" if data.historical_runs > 1 else "single incident" + if int(values.get(("ci", "fix_attempts"), "0")) > 0: + recs.append( + Recommendation( + "Move CI validation earlier", + "At least one autonomous CI repair was needed; reproduce the failing checks locally.", + ("ci.fix_attempts", "ci.failed_checks"), + scope, + ) + ) + if int(values.get(("review", "local_attempts"), "0")) > 1: + recs.append( + Recommendation( + "Strengthen pre-review validation", + "Multiple local review passes indicate an opportunity for an earlier validation gate.", + ("review.local_attempts",), + scope, + ) + ) + if data.outcome != "successful": + recs.append( + Recommendation( + "Review the terminal failure", + "The workflow did not complete successfully; inspect the bounded terminal evidence.", + ("workflow.blocked", "workflow.error", "workflow.terminal_node"), + scope, + ) + ) + return RetrospectiveReport(input=data, recommendations=tuple(recs)) + + +def format_report(report: RetrospectiveReport) -> str: + lines = [ + "## Forge retrospective", + "", + f"Outcome: **{report.input.outcome}**", + f"Evidence window: {len(report.input.evidence)} bounded facts; historical runs: " + f"{report.input.historical_runs}.", + "", + ] + if report.recommendations: + for rec in report.recommendations: + refs = ", ".join(f"`{ref}`" for ref in rec.evidence) + lines.extend([f"- **{rec.title}** ({rec.scope}): {rec.detail} Evidence: {refs}."]) + else: + lines.append("No actionable pattern was identified from the available evidence.") + lines.extend( + [ + "", + f"Analysis usage: {report.model}; input tokens={report.input_tokens}; " + f"output tokens={report.output_tokens}; estimated cost=${report.estimated_cost_usd:.4f}.", + ] + ) + return "\n".join(lines) + + +async def run_retrospective( + state: Mapping[str, Any], settings: Settings | None = None +) -> RetrospectiveReport | None: + """Publish a terminal report without changing state or terminal outcome.""" + settings = settings or get_settings() + is_terminal = state.get("current_node") == "complete" or bool(state.get("is_blocked")) + if ( + not settings.retrospective_enabled + or not is_terminal + or state.get("retrospective_completed", False) + ): + return None + data = build_input(state, settings.retrospective_max_items) + report = analyze(data) + jira = JiraClient() + await jira.add_comment(data.ticket_key, format_report(report)) + if settings.retrospective_create_issues: + await _create_recommendation_tasks(jira, report) + return report + + +async def _create_recommendation_tasks(jira: JiraClient, report: RetrospectiveReport) -> None: + """Create stable-keyed tasks; Jira labels provide idempotent deduplication.""" + project = report.input.ticket_key.split("-", 1)[0] + for rec in report.recommendations: + digest = hashlib.sha256(rec.title.encode()).hexdigest()[:12] + label = f"forge-retro-{digest}" + # Search is deliberately bounded to the target project and stable label. + matches = await jira.search_issues( + f'project = "{project}" AND labels = "{label}"', max_results=1 + ) + if matches: + continue + await jira.create_task( + project, + f"Forge retrospective: {rec.title}"[:255], + f"{rec.detail}\n\nEvidence: {', '.join(rec.evidence)}", + labels=["forge-retrospective", label], + ) diff --git a/tests/unit/workflow/test_retrospective.py b/tests/unit/workflow/test_retrospective.py new file mode 100644 index 000000000..fb9344e58 --- /dev/null +++ b/tests/unit/workflow/test_retrospective.py @@ -0,0 +1,130 @@ +from unittest.mock import AsyncMock, patch + +import pytest + +from forge.workflow.retrospective import ( + analyze, + build_input, + format_report, + run_retrospective, +) + + +def test_build_input_is_bounded_and_redacts_secrets(): + data = build_input( + { + "ticket_key": "F-1", + "current_node": "complete", + "last_error": "token=supersecret failed", + "ci_status": "failed", + "ci_fix_attempt": 2, + "ci_failed_checks": [{"name": str(i)} for i in range(50)], + }, + max_items=4, + ) + assert len(data.evidence) == 4 + assert "supersecret" not in repr(data) + assert data.outcome == "partially_completed" + + +@pytest.mark.parametrize( + ("state", "outcome"), + [ + ({"current_node": "complete"}, "successful"), + ({"current_node": "triage", "is_blocked": True}, "blocked"), + ({"current_node": "complete", "last_error": "partial"}, "partially_completed"), + ], +) +def test_outcomes(state, outcome): + assert build_input({"ticket_key": "F-1", **state}).outcome == outcome + + +def test_missing_observability_produces_valid_report(): + report = analyze(build_input({"ticket_key": "F-1", "current_node": "complete"})) + text = format_report(report) + assert "No actionable pattern" in text + assert "input tokens=0" in text + + +def test_report_links_recommendations_to_evidence(): + report = analyze( + build_input( + { + "ticket_key": "F-1", + "current_node": "complete", + "ci_fix_attempt": 2, + "ci_failed_checks": [{"name": "unit"}], + "local_review_attempts": 3, + } + ) + ) + text = format_report(report) + assert "`ci.fix_attempts`" in text + assert "`review.local_attempts`" in text + assert all(rec.scope == "single incident" for rec in report.recommendations) + + +@pytest.mark.asyncio +async def test_disabled_stage_has_no_side_effects(): + settings = AsyncMock(retrospective_enabled=False) + with patch("forge.workflow.retrospective.JiraClient") as jira: + assert await run_retrospective({"ticket_key": "F-1"}, settings) is None + jira.assert_not_called() + + +@pytest.mark.asyncio +async def test_nonterminal_stage_has_no_side_effects(): + settings = AsyncMock(retrospective_enabled=True) + with patch("forge.workflow.retrospective.JiraClient") as jira: + result = await run_retrospective( + {"ticket_key": "F-1", "current_node": "ci_evaluator"}, settings + ) + assert result is None + jira.assert_not_called() + + +@pytest.mark.asyncio +async def test_completed_stage_is_idempotent(): + settings = AsyncMock(retrospective_enabled=True) + with patch("forge.workflow.retrospective.JiraClient") as jira: + result = await run_retrospective( + {"ticket_key": "F-1", "retrospective_completed": True}, settings + ) + assert result is None + jira.assert_not_called() + + +@pytest.mark.asyncio +async def test_enabled_stage_publishes_comment_but_not_issues_by_default(): + settings = AsyncMock( + retrospective_enabled=True, + retrospective_create_issues=False, + retrospective_max_items=20, + ) + jira = AsyncMock() + with patch("forge.workflow.retrospective.JiraClient", return_value=jira): + result = await run_retrospective( + {"ticket_key": "F-1", "current_node": "complete", "ci_fix_attempt": 1}, + settings, + ) + assert result is not None + jira.add_comment.assert_awaited_once() + jira.create_task.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_issue_creation_is_deduplicated(): + settings = AsyncMock( + retrospective_enabled=True, + retrospective_create_issues=True, + retrospective_max_items=20, + ) + jira = AsyncMock() + jira.search_issues.return_value = [object()] + with patch("forge.workflow.retrospective.JiraClient", return_value=jira): + await run_retrospective( + {"ticket_key": "F-1", "current_node": "complete", "ci_fix_attempt": 1}, + settings, + ) + jira.search_issues.assert_awaited_once() + jira.create_task.assert_not_awaited()