')
+ html_parts.append('
')
-
- html_parts.append('Summary
') + html_parts.append(f'Specification: {spec_path}
')
+ html_parts.append(f'Run ID: {generation_id}
')
+ html_parts.append(
+ f'{variance.label}: '
+ f"{variance.value}
"
+ )
+ html_parts.append(
+ f'Final Estimate: '
+ f"{final_estimate_display}
"
+ )
+ html_parts.append(
+ f'Average: {average_display}
'
+ )
+ html_parts.append(
+ f'Range: {range_display}
'
+ )
+ if cost_display:
html_parts.append(
- 'π SpecFlow β Coding complete, starting deployment & QA
' - if pre_deploy - else 'β SpecFlow Iteration Complete
' + 'Total LLM cost (cumulative): '
+ f"{cost_display}
"
)
- html_parts.append('')
-
- # Summary section
+ html_parts.append('
')
+
+ # Variants section
+ if repo_links:
html_parts.append('')
- html_parts.append('report
Summary
') - html_parts.append(f'Specification: {spec_path}
')
- html_parts.append(f'Run ID: {generation_id}
')
- html_parts.append(
- f'{variance.label}: '
- f"{variance.value}
"
- )
- html_parts.append(
- f'Final Estimate: '
- f"{final_estimate_display}
"
- )
- html_parts.append(
- f'Average: {average_display}
'
- )
- html_parts.append(
- f'Range: {range_display}
'
- )
- if cost_display:
+ html_parts.append('Variants
') + html_parts.append('Click the links below to view the generation branches:
') + html_parts.append('')
+ for repo_link in repo_links:
+ html_parts.append('
')
+
+ html_parts.append('')
+
+ html_parts.append('')
+
+ html_content = '\n'.join(html_parts)
+
+ # Build plain text content
+ plain_parts = []
+ plain_parts.append("=" * 60)
+ plain_parts.append(
+ "SpecFlow β CODING COMPLETE, STARTING DEPLOYMENT & QA"
+ if pre_deploy
+ else "SpecFlow ITERATION COMPLETE"
+ )
+ plain_parts.append("=" * 60)
+ plain_parts.append("")
+ plain_parts.append(f"Specification: {spec_path}")
+ plain_parts.append(f"Run ID: {generation_id}")
+ plain_parts.append(f"{variance.label}: {variance.value}")
+ plain_parts.append(f"Final Estimate: {final_estimate_display}")
+ plain_parts.append(f"Average: {average_display}")
+ plain_parts.append(f"Range: {range_display}")
+ if cost_display:
+ plain_parts.append(f"Total LLM cost (cumulative): {cost_display}")
+ plain_parts.append("")
+
+ if repo_links:
+ plain_parts.append("VARIANTS:")
+ plain_parts.append("-" * 60)
+ for repo_link in repo_links:
+ plain_parts.append(f"{repo_link['workspace_id']}: {repo_link['branch_url']}")
+ model = repo_link.get("model_name")
+ codegen_lines = ws_codegen_lines.get(repo_link["workspace_id"], [])
+ if codegen_lines:
+ if model:
+ plain_parts.append(f" model: {model}")
+ for ln in codegen_lines:
+ plain_parts.append(f" {ln}")
+ else:
+ tokens_str = _format_tokens_millions(
+ repo_link.get("input_tokens"), repo_link.get("output_tokens")
+ )
+ if model or tokens_str:
+ detail_parts = []
if model:
- plain_parts.append(f" model: {model}")
- for ln in codegen_lines:
- plain_parts.append(f" {ln}")
- else:
- tokens_str = _format_tokens_millions(
- repo_link.get("input_tokens"), repo_link.get("output_tokens")
+ detail_parts.append(f"model: {model}")
+ if tokens_str:
+ detail_parts.append(f"tokens used: {tokens_str}")
+ plain_parts.append(f" {' | '.join(detail_parts)}")
+ llm_c = repo_link.get("total_usd_cost")
+ if llm_c is not None:
+ try:
+ plain_parts.append(
+ f" LLM API cost (cumulative): USD {float(llm_c):,.2f}"
)
- if model or tokens_str:
- detail_parts = []
- if model:
- detail_parts.append(f"model: {model}")
- if tokens_str:
- detail_parts.append(f"tokens used: {tokens_str}")
- plain_parts.append(f" {' | '.join(detail_parts)}")
- llm_c = repo_link.get("total_usd_cost")
- if llm_c is not None:
- try:
- plain_parts.append(
- f" LLM API cost (cumulative): USD {float(llm_c):,.2f}"
- )
- except (TypeError, ValueError):
- pass
- plain_parts.append("")
-
- if component_breakdown:
- plain_parts.append("COMPONENT BREAKDOWN:")
- plain_parts.append("-" * 60)
- # Build header
- header = "Component"
+ except (TypeError, ValueError):
+ pass
+ plain_parts.append("")
+
+ if component_breakdown:
+ plain_parts.append("COMPONENT BREAKDOWN:")
+ plain_parts.append("-" * 60)
+ # Build header
+ header = "Component"
+ if workspace_estimations:
+ for ws_est in workspace_estimations:
+ try:
+ ws_name = getattr(ws_est, 'workspace_name', 'unknown')
+ header += f" | {ws_name}"
+ except (AttributeError, TypeError):
+ continue
+ plain_parts.append(header)
+ plain_parts.append("-" * len(header))
+
+ for component_name, component_data in component_breakdown.items():
+ row = component_name
if workspace_estimations:
for ws_est in workspace_estimations:
try:
ws_name = getattr(ws_est, 'workspace_name', 'unknown')
- header += f" | {ws_name}"
- except (AttributeError, TypeError):
- continue
- plain_parts.append(header)
- plain_parts.append("-" * len(header))
-
- for component_name, component_data in component_breakdown.items():
- row = component_name
- if workspace_estimations:
- for ws_est in workspace_estimations:
- try:
- ws_name = getattr(ws_est, 'workspace_name', 'unknown')
- if ws_name in component_data["workspaces"]:
- hours = component_data["workspaces"][ws_name]["hours"]
- row += f" | {hours:.1f}"
- else:
- row += " | -"
- except (AttributeError, TypeError, KeyError):
+ if ws_name in component_data["workspaces"]:
+ hours = component_data["workspaces"][ws_name]["hours"]
+ row += f" | {hours:.1f}"
+ else:
row += " | -"
- plain_parts.append(row)
- plain_parts.append("")
-
- plain_parts.append(f"Generated at: {timestamp}")
- plain_content = '\n'.join(plain_parts)
-
- return html_content, plain_content
+ except (AttributeError, TypeError, KeyError):
+ row += " | -"
+ plain_parts.append(row)
+ plain_parts.append("")
+
+ plain_parts.append(f"Generated at: {timestamp}")
+ plain_content = '\n'.join(plain_parts)
+
+ return html_content, plain_content
# Initialize Notifications
diff --git a/backend/app/database/interface.py b/backend/app/database/interface.py
index cb1e091..3125c22 100644
--- a/backend/app/database/interface.py
+++ b/backend/app/database/interface.py
@@ -7,7 +7,7 @@
"""
from abc import ABC, abstractmethod
-from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar
+from typing import Any, Callable, Dict, List, Optional, Protocol, Tuple, TypeVar, runtime_checkable
# Type for query filter: (field, operator, value)
FilterTuple = Tuple[str, str, Any]
@@ -16,6 +16,22 @@
T = TypeVar("T")
+@runtime_checkable
+class ReadOnlyDatabase(Protocol):
+ """Read-only view over a database β the narrow subset a document reader needs.
+
+ Consumers that must never write status/checkpoint/workspace_phases (e.g. the
+ notifications report renderer) depend on this protocol instead of the full
+ ``IDatabase``, so those writes are impossible through the handle they hold.
+ Any ``IDatabase`` satisfies it structurally; ``StateMachineDBAdapter`` hands
+ out a view that exposes *only* ``get`` β keeping Commandment VII enforced,
+ not merely documented.
+ """
+
+ def get(self, collection: str, doc_id: str) -> Optional[Dict[str, Any]]:
+ ...
+
+
class ITransactionContext(ABC):
"""
Transaction context interface for atomic database operations.
diff --git a/backend/app/state/db_adapter.py b/backend/app/state/db_adapter.py
index 918e984..208e0cc 100644
--- a/backend/app/state/db_adapter.py
+++ b/backend/app/state/db_adapter.py
@@ -18,10 +18,10 @@
from __future__ import annotations
import asyncio
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any, Dict, Optional
if TYPE_CHECKING:
- from app.database.interface import IDatabase
+ from app.database.interface import IDatabase, ReadOnlyDatabase
# Firestore collection names β use these constants instead of repeating string literals
COL_WORKSPACES = "workspaces"
@@ -29,6 +29,24 @@
COL_API_KEYS = "api_keys"
+class _ReadOnlyDatabaseView:
+ """Read-only façade over ``IDatabase`` exposing only ``get``.
+
+ Handed to read-only consumers (the notifications report renderer) so they
+ physically cannot perform status/checkpoint/workspace_phases writes β the
+ encapsulation of Commandment VII is enforced by construction, not by a
+ docstring warning. Satisfies the ``ReadOnlyDatabase`` protocol structurally.
+ """
+
+ __slots__ = ("_db",)
+
+ def __init__(self, db: "IDatabase") -> None:
+ self._db = db
+
+ def get(self, collection: str, doc_id: str) -> Optional[Dict[str, Any]]:
+ return self._db.get(collection, doc_id)
+
+
class StateMachineDBAdapter:
"""
Adapts the synchronous IDatabase interface to the async interface
@@ -45,6 +63,18 @@ class StateMachineDBAdapter:
def __init__(self, db: "IDatabase") -> None:
self._db = db
+ @property
+ def read_only_db(self) -> "ReadOnlyDatabase":
+ """A read-only view over the underlying database.
+
+ For consumers that only read (e.g. the notifications report renderer,
+ which does a plain ``db.get("workspaces", id)``). The returned view
+ exposes *only* ``get`` β writes are unreachable through it, so
+ status/checkpoint/workspace_phases writes still have to go through this
+ adapter's async methods.
+ """
+ return _ReadOnlyDatabaseView(self._db)
+
# ------------------------------------------------------------------
# Generation methods
# ------------------------------------------------------------------
diff --git a/backend/app/workflows/multi_workspace_estimation_p10y.py b/backend/app/workflows/multi_workspace_estimation_p10y.py
index 787a390..e0d19fa 100644
--- a/backend/app/workflows/multi_workspace_estimation_p10y.py
+++ b/backend/app/workflows/multi_workspace_estimation_p10y.py
@@ -44,6 +44,8 @@
from app.schemas.specification import GenerationWorkflowRequest
from app.schemas.telemetry_workflow import TelemetryWorkflowLabel
from app.schemas.workspace import WorkspaceSettings
+from app.core.artifact_files import MULTI_WORKSPACE_REPORT_HTML_FILE, MULTI_WORKSPACE_REPORT_MD_FILE
+from app.core.notifications import render_generation_session_report_html
from app.services.artifact_store import ArtifactStore
from app.services.claude_code import agent_query
from app.services.p10y.estimation_report_generator import format_multi_workspace_report
@@ -642,7 +644,19 @@ async def multi_workspace_estimation_p10y_workflow(
f"buffer={risk_assessment.total_buffer_pct*100:.1f}%, "
f"final_estimate={risk_assessment.final_estimate:.1f}h"
)
-
+
+ # Build response now β every field is resolved, and both the markdown and
+ # HTML reports below render from it.
+ response = MultiWorkspaceEstimationResponse(
+ summary=summary,
+ workspace_estimations=workspace_estimations,
+ comparative_analysis=comparative_analysis,
+ timestamp=datetime.now(timezone.utc).isoformat(),
+ skipped_workspaces=skipped_workspaces,
+ aggregate_p10y_commit_coverage_pct=aggregate_cov,
+ total_usd_cost=session_llm_cost_total,
+ )
+
# Generate structured markdown report using report generator
logger.info("Generating structured markdown report...")
structured_report = format_multi_workspace_report(
@@ -652,13 +666,37 @@ async def multi_workspace_estimation_p10y_workflow(
skipped_workspaces=skipped_workspaces,
aggregate_p10y_commit_coverage_pct=aggregate_cov,
)
-
+
# Save structured report to file in primary workspace
primary_workspace_path = workspaces[0].workspace_path
full_outputs_dir = f"{primary_workspace_path}/{request.outputs_dir}"
- structured_report_path = f"{full_outputs_dir}/multi-workspace-estimation-report.md"
+ structured_report_path = f"{full_outputs_dir}/{MULTI_WORKSPACE_REPORT_MD_FILE}"
_write_structured_report(structured_report, structured_report_path, logger)
-
+
+ # Save the same report as HTML, regardless of whether email/Slack notifiers
+ # are configured β local quickstart users have neither, so this is the only
+ # place the HTML report is produced. Non-essential observability output: a
+ # failure here must never abort an estimation whose work is already done.
+ # The renderer only reads (a plain db.get("workspaces", id)); db_adapter is
+ # the async state-machine wrapper, so pass its read-only view β without a
+ # database handle the Variants section silently drops out.
+ logger.info("Generating HTML report...")
+ try:
+ html_content, _plain_content = render_generation_session_report_html(
+ generation_id=request.generation_id,
+ workspace_ids=workspace_ids,
+ result=response,
+ spec_path=request.spec_path,
+ db=db_adapter.read_only_db if db_adapter else None,
+ )
+ html_report_path = f"{full_outputs_dir}/{MULTI_WORKSPACE_REPORT_HTML_FILE}"
+ _write_structured_report(html_content, html_report_path, logger)
+ except Exception as e:
+ logger.warning(
+ "Failed to render/write HTML report β continuing without it: %s",
+ e, exc_info=True,
+ )
+
# Generate AI-powered comprehensive report (optional, additional analysis)
logger.info("Generating AI-powered comprehensive report...")
report_summary = await _generate_ai_report(
@@ -691,18 +729,7 @@ async def multi_workspace_estimation_p10y_workflow(
f"Failed to archive combined report for estimation {generation_id}: {e}",
exc_info=True,
)
-
- # Build response
- response = MultiWorkspaceEstimationResponse(
- summary=summary,
- workspace_estimations=workspace_estimations,
- comparative_analysis=comparative_analysis,
- timestamp=datetime.now(timezone.utc).isoformat(),
- skipped_workspaces=skipped_workspaces,
- aggregate_p10y_commit_coverage_pct=aggregate_cov,
- total_usd_cost=session_llm_cost_total,
- )
-
+
logger.info("Multi-workspace estimation completed successfully")
-
+
return response
diff --git a/backend/scripts/example_test_session.py b/backend/scripts/example_test_session.py
new file mode 100644
index 0000000..6f29afa
--- /dev/null
+++ b/backend/scripts/example_test_session.py
@@ -0,0 +1,248 @@
+#!/usr/bin/env python3
+"""
+Seed one fake, already-COMPLETED generation session for local TUI testing.
+
+Writes a generation_sessions document (with a full P10Y result β summary,
+per-workspace breakdown, component comparison) plus the markdown/HTML reports
+on disk, so `specflow tui` can render a finished run without waiting for a
+real generation.
+
+Requires the local sentinel identity to already exist (api_keys/local) β
+run `specflow init` (or `python scripts/init_firestore.py`) at least once
+before this script.
+"""
+
+import argparse
+import os
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+# Set DATABASE_TYPE early if FIRESTORE_EMULATOR_HOST is set (before importing settings)
+if os.getenv("FIRESTORE_EMULATOR_HOST") and not os.getenv("DATABASE_TYPE"):
+ os.environ["DATABASE_TYPE"] = "emulator"
+
+from app.core.artifact_files import MULTI_WORKSPACE_REPORT_HTML_FILE, MULTI_WORKSPACE_REPORT_MD_FILE
+from app.core.artifact_subdirs import REPORT_SUBDIR
+from app.core.local_identity import LOCAL_API_KEY_DOC_ID
+from app.core.notifications import render_generation_session_report_html
+from app.core.workspace_pool_names import DEFAULT_WORKSPACE_POOL
+from app.database.factory import get_database
+from app.schemas.estimate import (
+ ComparativeAnalysis,
+ ComponentComparison,
+ ComponentEstimation,
+ EstimationMetrics,
+ EstimationSummary,
+ MultiWorkspaceEstimationResponse,
+ RiskAssessment,
+ WorkspaceEstimation,
+)
+from app.schemas.generation_workflow_enums import GenerationCheckpoint, GenerationStatus
+from app.services.artifact_store import ARTIFACTS_BASE
+from app.services.p10y.estimation_report_generator import format_multi_workspace_report
+from app.state.db_adapter import COL_GENERATION_SESSIONS
+
+# Matches tui.constants.LOCAL_ONLY_READINESS β kept as a literal since the TUI
+# package isn't importable from the backend.
+LOCAL_ONLY_READINESS = "LOCAL_ONLY"
+
+COMPONENTS = [
+ ("auth", 0.40, 6.0),
+ ("billing", 0.35, 9.5),
+ ("api", 0.25, 3.2),
+]
+
+
+def _workspace_estimation(workspace_name: str, total_hours: float) -> WorkspaceEstimation:
+ component_breakdown = {
+ name: ComponentEstimation(
+ component_name=name,
+ hours=round(total_hours * share, 1),
+ new_work=round(total_hours * share * 0.7, 1),
+ refactor=round(total_hours * share * 0.25, 1),
+ rework=round(total_hours * share * 0.05, 1),
+ quality_score=0.9,
+ )
+ for name, share, _variance in COMPONENTS
+ }
+ return WorkspaceEstimation(
+ workspace_name=workspace_name,
+ workspace_path=f"/workspaces/{workspace_name}",
+ total_hours=total_hours,
+ total_effective_output=round(total_hours * 0.9, 1),
+ component_breakdown=component_breakdown,
+ estimation_metrics=EstimationMetrics(
+ new_work=round(total_hours * 0.7, 1),
+ refactor=round(total_hours * 0.25, 1),
+ rework=round(total_hours * 0.05, 1),
+ removed_work=0.0,
+ quality_score=0.9,
+ effective_output=round(total_hours * 0.9, 1),
+ total_output=total_hours,
+ ),
+ commits_count=12,
+ total_usd_cost=4.25,
+ )
+
+
+def build_estimation_result(workspace_ids: list[str]) -> MultiWorkspaceEstimationResponse:
+ """Fake P10Y result with a real component breakdown, built from the actual schemas."""
+ hours_by_ws = [110.0 + 5.0 * i for i in range(len(workspace_ids))]
+ workspace_estimations = [
+ _workspace_estimation(ws_id, hours) for ws_id, hours in zip(workspace_ids, hours_by_ws)
+ ]
+
+ average_hours = sum(hours_by_ws) / len(hours_by_ws)
+ summary = EstimationSummary(
+ average_hours=average_hours,
+ std_deviation=3.3,
+ min_hours=min(hours_by_ws),
+ max_hours=max(hours_by_ws),
+ coefficient_of_variation=0.03,
+ variance_assessment="low",
+ risk_assessment=RiskAssessment(
+ status="Approved",
+ instability_ratio=0.03,
+ rejection_threshold=0.15,
+ base_component=0.10,
+ var_component=0.02,
+ size_component=0.01,
+ total_buffer_pct=0.13,
+ final_estimate=average_hours * 1.13,
+ ),
+ )
+
+ component_comparison = {
+ name: ComponentComparison(
+ component_name=name,
+ hours_by_workspace={
+ ws.workspace_name: ws.component_breakdown[name].hours for ws in workspace_estimations
+ },
+ average=sum(ws.component_breakdown[name].hours for ws in workspace_estimations)
+ / len(workspace_estimations),
+ std_deviation=2.1,
+ variance_percentage=variance,
+ )
+ for name, _share, variance in COMPONENTS
+ }
+ comparative_analysis = ComparativeAnalysis(
+ component_comparison=component_comparison,
+ high_variance_components=["billing"],
+ insights=[
+ "Low variance across workspaces",
+ "Billing has the highest variance β consider a closer look",
+ ],
+ )
+
+ return MultiWorkspaceEstimationResponse(
+ summary=summary,
+ workspace_estimations=workspace_estimations,
+ comparative_analysis=comparative_analysis,
+ timestamp=datetime.now(timezone.utc).isoformat(),
+ total_usd_cost=sum(ws.total_usd_cost for ws in workspace_estimations),
+ )
+
+
+def write_reports(generation_id: str, result: MultiWorkspaceEstimationResponse, spec_path: str) -> Path:
+ """Write the markdown + HTML reports where the report.html endpoint expects to find them."""
+ report_dir = ARTIFACTS_BASE / generation_id / REPORT_SUBDIR
+ report_dir.mkdir(parents=True, exist_ok=True)
+
+ markdown = format_multi_workspace_report(
+ workspace_estimations=result.workspace_estimations,
+ summary=result.summary,
+ comparative_analysis=result.comparative_analysis,
+ skipped_workspaces=result.skipped_workspaces,
+ aggregate_p10y_commit_coverage_pct=result.aggregate_p10y_commit_coverage_pct,
+ )
+ (report_dir / MULTI_WORKSPACE_REPORT_MD_FILE).write_text(markdown)
+
+ html_content, _plain = render_generation_session_report_html(
+ generation_id=generation_id,
+ workspace_ids=[ws.workspace_name for ws in result.workspace_estimations],
+ result=result,
+ spec_path=spec_path,
+ db=None,
+ )
+ (report_dir / MULTI_WORKSPACE_REPORT_HTML_FILE).write_text(html_content)
+
+ return report_dir
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument(
+ "--generation-id",
+ default="est-example0001",
+ help="Generation ID to seed (default: est-example0001). Overwrites if it already exists.",
+ )
+ parser.add_argument(
+ "--spec-path",
+ default="specs/example.md",
+ help="Fake spec path stored on the session (cosmetic only).",
+ )
+ args = parser.parse_args()
+
+ db = get_database()
+
+ sentinel = db.get("api_keys", LOCAL_API_KEY_DOC_ID)
+ if not sentinel:
+ print(
+ f"ERROR: local sentinel identity (api_keys/{LOCAL_API_KEY_DOC_ID}) not found.\n"
+ "Run `specflow init` (or `python scripts/init_firestore.py`) at least once first."
+ )
+ return 1
+ user_email = (sentinel.get("user_id") or "").lower()
+ key_uid = sentinel.get("key_uid")
+ workspace_pool = sentinel.get("workspace_pool") or DEFAULT_WORKSPACE_POOL
+
+ workspace_ids = ["ws-01-1", "ws-01-2", "ws-01-3"]
+ result = build_estimation_result(workspace_ids)
+ report_dir = write_reports(args.generation_id, result, args.spec_path)
+
+ now = datetime.now(timezone.utc)
+ workspace_phases = {
+ ws_id: {"last_completed_phase": 9, "total_phases": 9, "phase_name": "Done"}
+ for ws_id in workspace_ids
+ }
+
+ db.set(
+ COL_GENERATION_SESSIONS,
+ args.generation_id,
+ {
+ "generation_id": args.generation_id,
+ "user_email": user_email,
+ "key_uid": key_uid,
+ "workspace_pool": workspace_pool,
+ "status": GenerationStatus.COMPLETED.value,
+ "checkpoint": GenerationCheckpoint.ESTIMATION_DONE.value,
+ "created_at": now,
+ "started_at": now,
+ "completed_at": now,
+ "workspace_ids": workspace_ids,
+ "parameters": {
+ "workspace_count": len(workspace_ids),
+ "spec_path": args.spec_path,
+ "outputs_dir": "docs",
+ },
+ "last_spec_readiness": LOCAL_ONLY_READINESS,
+ "workspace_phases": workspace_phases,
+ "progress": {},
+ "result": result.model_dump(),
+ "artifact_path": str(ARTIFACTS_BASE / args.generation_id),
+ "code_archived": True,
+ "retry_count": 0,
+ "max_retries": 3,
+ "error": None,
+ },
+ )
+
+ print(f"Seeded completed generation session: {args.generation_id}")
+ print(f"Reports written to: {report_dir}")
+ print(f"\nOn your host machine, run:\n\n specflow tui --generation-id {args.generation_id}\n")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/backend/test/api/test_email_notifications.py b/backend/test/api/test_email_notifications.py
index 97adc9e..f522c34 100644
--- a/backend/test/api/test_email_notifications.py
+++ b/backend/test/api/test_email_notifications.py
@@ -545,6 +545,60 @@ def test_notify_generation_session_complete_with_missing_workspace(
# But email should still be sent successfully
+class TestRenderGenerationSessionReportHtml:
+ """``render_generation_session_report_html`` is usable with no SMTP/EmailConfig β
+ this is the function the P10Y workflow calls to save the report to disk even
+ when no notifier is configured (local quickstart)."""
+
+ def test_builds_html_and_plain_without_any_email_config(
+ self, mock_db, sample_workspace_docs, sample_estimation_result
+ ):
+ from app.core.notifications import render_generation_session_report_html
+
+ workspace_ids, _ = sample_workspace_docs
+
+ html_content, plain_content = render_generation_session_report_html(
+ generation_id="est-test-123",
+ workspace_ids=workspace_ids,
+ result=sample_estimation_result,
+ spec_path="specs/test.md",
+ db=mock_db,
+ )
+
+ assert html_content.strip() != ""
+ assert plain_content.strip() != ""
+ assert "')
+ html_parts.append(f'{repo_link["workspace_id"]}: ')
html_parts.append(
- '
')
-
- # Variants section
- if repo_links:
- html_parts.append('Total LLM cost (cumulative): '
- f"{cost_display}
"
+ f'Branch link'
)
- html_parts.append('')
- html_parts.append('
' + f"LLM API cost (cumulative): USD {float(llm_c):,.2f}" + ) + except (TypeError, ValueError): + pass html_parts.append('
')
-
- # Component breakdown section
- if component_breakdown:
- html_parts.append('Variants
') - html_parts.append('Click the links below to view the generation branches:
') - html_parts.append('')
- for repo_link in repo_links:
- html_parts.append('
')
+ llm_c = repo_link.get("total_usd_cost")
+ if llm_c is not None:
+ try:
+ html_parts.append(
+ f'')
- html_parts.append(f'{repo_link["workspace_id"]}: ')
- html_parts.append(
- f'Branch link'
+ model = repo_link.get("model_name")
+ codegen_detail = ws_codegen_lines.get(repo_link["workspace_id"], [])
+ if codegen_detail:
+ html_parts.append('
') + if model: + html_parts.append(f"model: {model}
") + html_parts.append("
".join(codegen_detail)) + html_parts.append("") + else: + tokens_str = _format_tokens_millions( + repo_link.get("input_tokens"), repo_link.get("output_tokens") ) - model = repo_link.get("model_name") - codegen_detail = ws_codegen_lines.get(repo_link["workspace_id"], []) - if codegen_detail: - html_parts.append('
') + if model or tokens_str: + detail_parts = [] if model: - html_parts.append(f"model: {model}
") - html_parts.append("
".join(codegen_detail)) - html_parts.append("") - else: - tokens_str = _format_tokens_millions( - repo_link.get("input_tokens"), repo_link.get("output_tokens") + detail_parts.append(f"model: {model}") + if tokens_str: + detail_parts.append(f"tokens used: {tokens_str}") + html_parts.append( + f'
' + f'{" | ".join(detail_parts)}' ) - if model or tokens_str: - detail_parts = [] - if model: - detail_parts.append(f"model: {model}") - if tokens_str: - detail_parts.append(f"tokens used: {tokens_str}") - html_parts.append( - f'
' - f'{" | ".join(detail_parts)}' - ) - llm_c = repo_link.get("total_usd_cost") - if llm_c is not None: - try: - html_parts.append( - f'
' - f"LLM API cost (cumulative): USD {float(llm_c):,.2f}" - ) - except (TypeError, ValueError): - pass - html_parts.append('
')
- html_parts.append('') + if model: + html_parts.append(f"model: {model}
") + html_parts.append("
".join(codegen_detail)) + html_parts.append("") + else: + tokens_str = _format_tokens_millions( + repo_link.get("input_tokens"), repo_link.get("output_tokens") ) - model = repo_link.get("model_name") - codegen_detail = ws_codegen_lines.get(repo_link["workspace_id"], []) - if codegen_detail: - html_parts.append('
') + if model or tokens_str: + detail_parts = [] if model: - html_parts.append(f"model: {model}
") - html_parts.append("
".join(codegen_detail)) - html_parts.append("") - else: - tokens_str = _format_tokens_millions( - repo_link.get("input_tokens"), repo_link.get("output_tokens") + detail_parts.append(f"model: {model}") + if tokens_str: + detail_parts.append(f"tokens used: {tokens_str}") + html_parts.append( + f'
' + f'{" | ".join(detail_parts)}' ) - if model or tokens_str: - detail_parts = [] - if model: - detail_parts.append(f"model: {model}") - if tokens_str: - detail_parts.append(f"tokens used: {tokens_str}") - html_parts.append( - f'
' - f'{" | ".join(detail_parts)}' - ) - llm_c = repo_link.get("total_usd_cost") - if llm_c is not None: - try: - html_parts.append( - f'
' - f"LLM API cost (cumulative): USD {float(llm_c):,.2f}" - ) - except (TypeError, ValueError): - pass - html_parts.append('
' + f"LLM API cost (cumulative): USD {float(llm_c):,.2f}" + ) + except (TypeError, ValueError): + pass html_parts.append('
')
- html_parts.append('')
- html_parts.append('
')
html_parts.append('
')
-
- html_parts.append('')
-
- html_content = '\n'.join(html_parts)
-
- # Build plain text content
- plain_parts = []
- plain_parts.append("=" * 60)
- plain_parts.append(
- "SpecFlow β CODING COMPLETE, STARTING DEPLOYMENT & QA"
- if pre_deploy
- else "SpecFlow ITERATION COMPLETE"
- )
- plain_parts.append("=" * 60)
- plain_parts.append("")
- plain_parts.append(f"Specification: {spec_path}")
- plain_parts.append(f"Run ID: {generation_id}")
- plain_parts.append(f"{variance.label}: {variance.value}")
- plain_parts.append(f"Final Estimate: {final_estimate_display}")
- plain_parts.append(f"Average: {average_display}")
- plain_parts.append(f"Range: {range_display}")
- if cost_display:
- plain_parts.append(f"Total LLM cost (cumulative): {cost_display}")
- plain_parts.append("")
-
- if repo_links:
- plain_parts.append("VARIANTS:")
- plain_parts.append("-" * 60)
- for repo_link in repo_links:
- plain_parts.append(f"{repo_link['workspace_id']}: {repo_link['branch_url']}")
- model = repo_link.get("model_name")
- codegen_lines = ws_codegen_lines.get(repo_link["workspace_id"], [])
- if codegen_lines:
+
+ html_parts.append('Component Complexity Metrics Breakdown
') - html_parts.append('| Component | ') - # Add workspace columns + html_parts.append('') + html_parts.append('') + + # Component breakdown section + if component_breakdown: + html_parts.append('
|---|
| Component | ') + # Add workspace columns + if workspace_estimations: + for ws_est in workspace_estimations: + try: + ws_name = getattr(ws_est, 'workspace_name', 'unknown') + html_parts.append(f'{ws_name} (hours) | ')
+ except (AttributeError, TypeError):
+ continue
+ html_parts.append('|||
|---|---|---|---|---|
| {component_name} | ') + # Add hours for each workspace if workspace_estimations: for ws_est in workspace_estimations: try: ws_name = getattr(ws_est, 'workspace_name', 'unknown') - html_parts.append(f'{ws_name} (hours) | ')
- except (AttributeError, TypeError):
- continue
- html_parts.append('|||
| {component_name} | ') - # Add hours for each workspace - if workspace_estimations: - for ws_est in workspace_estimations: - try: - ws_name = getattr(ws_est, 'workspace_name', 'unknown') - if ws_name in component_data["workspaces"]: - hours = component_data["workspaces"][ws_name]["hours"] - html_parts.append(f'{hours:.1f} | ') - else: - html_parts.append('- | ') - except (AttributeError, TypeError, KeyError): + if ws_name in component_data["workspaces"]: + hours = component_data["workspaces"][ws_name]["hours"] + html_parts.append(f'{hours:.1f} | ') + else: html_parts.append('- | ') - html_parts.append('