# [DEF:backend.src.services.reports.report_service:Module] # @TIER: CRITICAL # @SEMANTICS: reports, service, aggregation, filtering, pagination, detail # @PURPOSE: Aggregate, normalize, filter, and paginate task reports for unified list/detail API use cases. # @LAYER: Domain # @RELATION: DEPENDS_ON -> backend.src.core.task_manager.manager.TaskManager # @RELATION: DEPENDS_ON -> backend.src.models.report # @RELATION: DEPENDS_ON -> backend.src.services.reports.normalizer # @INVARIANT: List responses are deterministic and include applied filter echo metadata. # [SECTION: IMPORTS] from datetime import datetime, timezone from typing import List, Optional from ...core.logger import belief_scope from ...core.task_manager import TaskManager from ...models.report import ReportCollection, ReportDetailView, ReportQuery, ReportStatus, TaskReport, TaskType from .normalizer import normalize_task_report # [/SECTION] # [DEF:ReportsService:Class] # @PURPOSE: Service layer for list/detail report retrieval and normalization. # @TIER: CRITICAL # @PRE: TaskManager dependency is initialized. # @POST: Provides deterministic list/detail report responses. # @INVARIANT: Service methods are read-only over task history source. class ReportsService: # [DEF:__init__:Function] # @TIER: CRITICAL # @PURPOSE: Initialize service with TaskManager dependency. # @PRE: task_manager is a live TaskManager instance. # @POST: self.task_manager is assigned and ready for read operations. # @INVARIANT: Constructor performs no task mutations. # @PARAM: task_manager (TaskManager) - Task manager providing source task history. def __init__(self, task_manager: TaskManager): with belief_scope("__init__"): self.task_manager = task_manager # [/DEF:__init__:Function] # [DEF:_load_normalized_reports:Function] # @PURPOSE: Build normalized reports from all available tasks. # @PRE: Task manager returns iterable task history records. # @POST: Returns normalized report list preserving source cardinality. # @INVARIANT: Every returned item is a TaskReport. # @RETURN: List[TaskReport] - Reports sorted later by list logic. def _load_normalized_reports(self) -> List[TaskReport]: with belief_scope("_load_normalized_reports"): tasks = self.task_manager.get_all_tasks() reports = [normalize_task_report(task) for task in tasks] return reports # [/DEF:_load_normalized_reports:Function] # [DEF:_to_utc_datetime:Function] # @PURPOSE: Normalize naive/aware datetime values to UTC-aware datetime for safe comparisons. # @PRE: value is either datetime or None. # @POST: Returns UTC-aware datetime or None. # @INVARIANT: Naive datetimes are interpreted as UTC to preserve deterministic ordering/filtering. # @PARAM: value (Optional[datetime]) - Source datetime value. # @RETURN: Optional[datetime] - UTC-aware datetime or None. def _to_utc_datetime(self, value: Optional[datetime]) -> Optional[datetime]: with belief_scope("_to_utc_datetime"): if value is None: return None if value.tzinfo is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) # [/DEF:_to_utc_datetime:Function] # [DEF:_datetime_sort_key:Function] # @PURPOSE: Produce stable numeric sort key for report timestamps. # @PRE: report contains updated_at datetime. # @POST: Returns float timestamp suitable for deterministic sorting. # @INVARIANT: Mixed naive/aware datetimes never raise TypeError. # @PARAM: report (TaskReport) - Report item. # @RETURN: float - UTC timestamp key. def _datetime_sort_key(self, report: TaskReport) -> float: with belief_scope("_datetime_sort_key"): updated = self._to_utc_datetime(report.updated_at) if updated is None: return 0.0 return updated.timestamp() # [/DEF:_datetime_sort_key:Function] # [DEF:_matches_query:Function] # @PURPOSE: Apply query filtering to a report. # @PRE: report and query are normalized schema instances. # @POST: Returns True iff report satisfies all active query filters. # @INVARIANT: Filter evaluation is side-effect free. # @PARAM: report (TaskReport) - Candidate report. # @PARAM: query (ReportQuery) - Applied query. # @RETURN: bool - True if report matches all filters. def _matches_query(self, report: TaskReport, query: ReportQuery) -> bool: with belief_scope("_matches_query"): if query.task_types and report.task_type not in query.task_types: return False if query.statuses and report.status not in query.statuses: return False report_updated_at = self._to_utc_datetime(report.updated_at) query_time_from = self._to_utc_datetime(query.time_from) query_time_to = self._to_utc_datetime(query.time_to) if query_time_from and report_updated_at and report_updated_at < query_time_from: return False if query_time_to and report_updated_at and report_updated_at > query_time_to: return False if query.search: needle = query.search.lower() haystack = f"{report.summary} {report.task_type.value} {report.status.value}".lower() if needle not in haystack: return False return True # [/DEF:_matches_query:Function] # [DEF:_sort_reports:Function] # @PURPOSE: Sort reports deterministically according to query settings. # @PRE: reports contains only TaskReport items. # @POST: Returns reports ordered by selected sort field and order. # @INVARIANT: Sorting criteria are deterministic for equal input. # @PARAM: reports (List[TaskReport]) - Filtered reports. # @PARAM: query (ReportQuery) - Sort config. # @RETURN: List[TaskReport] - Sorted reports. def _sort_reports(self, reports: List[TaskReport], query: ReportQuery) -> List[TaskReport]: with belief_scope("_sort_reports"): reverse = query.sort_order == "desc" if query.sort_by == "status": reports.sort(key=lambda item: item.status.value, reverse=reverse) elif query.sort_by == "task_type": reports.sort(key=lambda item: item.task_type.value, reverse=reverse) else: reports.sort(key=self._datetime_sort_key, reverse=reverse) return reports # [/DEF:_sort_reports:Function] # [DEF:list_reports:Function] # @PURPOSE: Return filtered, sorted, paginated report collection. # @PRE: query has passed schema validation. # @POST: Returns {items,total,page,page_size,has_next,applied_filters}. # @PARAM: query (ReportQuery) - List filters and pagination. # @RETURN: ReportCollection - Paginated unified reports payload. def list_reports(self, query: ReportQuery) -> ReportCollection: with belief_scope("list_reports"): reports = self._load_normalized_reports() filtered = [report for report in reports if self._matches_query(report, query)] sorted_reports = self._sort_reports(filtered, query) total = len(sorted_reports) start = (query.page - 1) * query.page_size end = start + query.page_size items = sorted_reports[start:end] has_next = end < total return ReportCollection( items=items, total=total, page=query.page, page_size=query.page_size, has_next=has_next, applied_filters=query, ) # [/DEF:list_reports:Function] # [DEF:get_report_detail:Function] # @PURPOSE: Return one normalized report with timeline/diagnostics/next actions. # @PRE: report_id exists in normalized report set. # @POST: Returns normalized detail envelope with diagnostics and next actions where applicable. # @PARAM: report_id (str) - Stable report identifier. # @RETURN: Optional[ReportDetailView] - Detailed report or None if not found. def get_report_detail(self, report_id: str) -> Optional[ReportDetailView]: with belief_scope("get_report_detail"): reports = self._load_normalized_reports() target = next((report for report in reports if report.report_id == report_id), None) if not target: return None timeline = [] if target.started_at: timeline.append({"event": "started", "at": target.started_at.isoformat()}) timeline.append({"event": "updated", "at": target.updated_at.isoformat()}) diagnostics = target.details or {} if not diagnostics: diagnostics = {"note": "Not provided"} if target.error_context: diagnostics["error_context"] = target.error_context.model_dump() next_actions = [] if target.error_context and target.error_context.next_actions: next_actions = target.error_context.next_actions elif target.status in {ReportStatus.FAILED, ReportStatus.PARTIAL}: next_actions = ["Review diagnostics", "Retry task if applicable"] return ReportDetailView( report=target, timeline=timeline, diagnostics=diagnostics, next_actions=next_actions, ) # [/DEF:get_report_detail:Function] # [/DEF:ReportsService:Class] import sys from generate_semantic_map import parse_file file_path = "backend/src/core/task_manager/task_logger.py" entities, issues = parse_file(file_path, file_path, "python") for e in entities: e.validate() def print_entity(ent, indent=0): print(" " * indent + f"{ent.type} {ent.name} Tags: {list(ent.tags.keys())} Belief: {ent.has_belief_scope}") for i in ent.compliance_issues: print(" " * (indent + 1) + f"ISSUE: {i.message}") for c in ent.children: print_entity(c, indent + 1) for e in entities: print_entity(e) for i in issues: print(f"GLOBAL ISSUE: {i.message} at line {i.line_number}") # [/DEF:backend.src.services.reports.report_service:Module]