test contracts

backend/src/api/routes/reports.py

@@ -62,6 +62,20 @@ def _parse_csv_enum_list(raw: Optional[str], enum_cls, field_name: str) -> List:
 # @PRE: authenticated/authorized request and validated query params.
 # @POST: returns {items,total,page,page_size,has_next,applied_filters}.
 # @POST: deterministic error payload for invalid filters.
+#
+# @TEST_CONTRACT: ListReportsApi ->
+# {
+#   required_fields: {page: int, page_size: int, sort_by: str, sort_order: str},
+#   optional_fields: {task_types: str, statuses: str, search: str},
+#   invariants: [
+#       "Returns ReportCollection on success",
+#       "Raises HTTPException 400 for invalid query parameters"
+#   ]
+# }
+# @TEST_FIXTURE: valid_list_request -> {"page": 1, "page_size": 20}
+# @TEST_EDGE: invalid_task_type_filter -> raises HTTPException(400)
+# @TEST_EDGE: malformed_query -> raises HTTPException(400)
+# @TEST_INVARIANT: consistent_list_payload -> verifies: [valid_list_request]
 @router.get("", response_model=ReportCollection)
 async def list_reports(
     page: int = Query(1, ge=1),

backend/src/app.py

@@ -147,6 +147,21 @@ app.include_router(assistant.router, prefix="/api/assistant", tags=["Assistant"]
 # @POST: WebSocket connection is managed and logs are streamed until disconnect.
 # @TIER: CRITICAL
 # @UX_STATE: Connecting -> Streaming -> (Disconnected)
+#
+# @TEST_CONTRACT: WebSocketLogStreamApi ->
+# {
+#   required_fields: {websocket: WebSocket, task_id: str},
+#   optional_fields: {source: str, level: str},
+#   invariants: [
+#       "Accepts the WebSocket connection",
+#       "Applies source and level filters correctly to streamed logs",
+#       "Cleans up subscriptions on disconnect"
+#   ]
+# }
+# @TEST_FIXTURE: valid_ws_connection -> {"task_id": "test_1", "source": "plugin"}
+# @TEST_EDGE: task_not_found_ws -> closes connection or sends error
+# @TEST_EDGE: empty_task_logs -> waits for new logs
+# @TEST_INVARIANT: consistent_streaming -> verifies: [valid_ws_connection]
 @app.websocket("/ws/logs/{task_id}")
 async def websocket_endpoint(
     websocket: WebSocket,

backend/src/core/mapping_service.py

@@ -23,6 +23,21 @@ from src.core.logger import logger, belief_scope
 # [DEF:IdMappingService:Class]
 # @TIER: CRITICAL
 # @PURPOSE: Service handling the cataloging and retrieval of remote Superset Integer IDs.
+#
+# @TEST_CONTRACT: IdMappingServiceModel ->
+# {
+#   required_fields: {db_session: Session},
+#   invariants: [
+#       "sync_environment correctly creates or updates ResourceMapping records",
+#       "get_remote_id returns an integer or None",
+#       "get_remote_ids_batch returns a dictionary of valid UUIDs to integers"
+#   ]
+# }
+# @TEST_FIXTURE: valid_mapping_service -> {"db_session": "MockSession()"}
+# @TEST_EDGE: sync_api_failure -> handles exception gracefully
+# @TEST_EDGE: get_remote_id_not_found -> returns None
+# @TEST_EDGE: get_batch_empty_list -> returns empty dict
+# @TEST_INVARIANT: resilient_fetching -> verifies: [sync_api_failure]
 class IdMappingService:
 
     # [DEF:__init__:Function]

backend/src/core/task_manager/context.py

@@ -19,6 +19,20 @@ from ..logger import belief_scope
 # @TIER: CRITICAL
 # @INVARIANT: logger is always a valid TaskLogger instance.
 # @UX_STATE: Idle -> Active -> Complete
+#
+# @TEST_CONTRACT: TaskContextInit ->
+# {
+#   required_fields: {task_id: str, add_log_fn: Callable, params: dict},
+#   optional_fields: {default_source: str},
+#   invariants: [
+#       "task_id matches initialized logger's task_id",
+#       "logger is a valid TaskLogger instance"
+#   ]
+# }
+# @TEST_FIXTURE: valid_context -> {"task_id": "123", "add_log_fn": lambda *args: None, "params": {"k": "v"}, "default_source": "plugin"}
+# @TEST_EDGE: missing_task_id -> raises TypeError
+# @TEST_EDGE: missing_add_log_fn -> raises TypeError
+# @TEST_INVARIANT: logger_initialized -> verifies: [valid_context]
 class TaskContext:
     """
     Execution context provided to plugins during task execution.

backend/src/core/task_manager/manager.py

@@ -39,17 +39,20 @@ from ..logger import logger, belief_scope, should_log_task_level
 # @INVARIANT: Task IDs are unique within the registry.
 # @INVARIANT: Each task has exactly one status at any time.
 # @INVARIANT: Log entries are never deleted after being added to a task.
-# @TEST_CONTRACT: TaskManager -> {
-#    required_fields: {plugin_id: str, params: dict},
-#    optional_fields: {user_id: str},
-#    invariants: ["Task IDs are unique within the registry", "Each task has exactly one status at any time"]
+#
+# @TEST_CONTRACT: TaskManagerModel ->
+# {
+#   required_fields: {plugin_loader: PluginLoader},
+#   invariants: [
+#       "Tasks are persisted immediately upon creation",
+#       "Running tasks use a thread pool or asyncio event loop based on executor type",
+#       "Log flushing runs on a background thread"
+#   ]
 # }
-# @TEST_FIXTURE: create_valid_task -> {"plugin_id": "migration_plugin", "params": {"source": "A", "target": "B"}}
-# @TEST_EDGE: missing_required_field -> {"plugin_id": null}
-# @TEST_EDGE: empty_response -> {"params": {}}
-# @TEST_EDGE: invalid_type -> {"params": "string_instead_of_dict"}
-# @TEST_EDGE: external_failure -> {"plugin_not_found": true}
-# @TEST_INVARIANT: single_status -> verifies: [create_valid_task, external_failure]
+# @TEST_FIXTURE: valid_manager -> {"plugin_loader": "MockPluginLoader()"}
+# @TEST_EDGE: create_task_invalid_plugin -> raises ValueError
+# @TEST_EDGE: create_task_invalid_params -> raises ValueError
+# @TEST_INVARIANT: lifecycle_management -> verifies: [valid_manager]
     """
     Manages the lifecycle of tasks, including their creation, execution, and state tracking.
     """

backend/src/core/task_manager/models.py

@@ -45,6 +45,14 @@ class LogLevel(str, Enum):
 # @PURPOSE: A Pydantic model representing a single, structured log entry associated with a task.
 # @TIER: CRITICAL
 # @INVARIANT: Each log entry has a unique timestamp and source.
+#
+# @TEST_CONTRACT: LogEntryModel ->
+# {
+#   required_fields: {message: str},
+#   optional_fields: {timestamp: datetime, level: str, source: str, context: dict, metadata: dict}
+# }
+# @TEST_FIXTURE: valid_log_entry -> {"message": "Plugin initialized"}
+# @TEST_EDGE: empty_message -> {"message": ""}
 class LogEntry(BaseModel):
     timestamp: datetime = Field(default_factory=datetime.utcnow)
     level: str = Field(default="INFO")

backend/src/core/task_manager/persistence.py

@@ -24,6 +24,20 @@ from ..logger import logger, belief_scope
 # @SEMANTICS: persistence, service, database, sqlalchemy
 # @PURPOSE: Provides methods to save and load tasks from the tasks.db database using SQLAlchemy.
 # @INVARIANT: Persistence must handle potentially missing task fields natively.
+#
+# @TEST_CONTRACT: TaskPersistenceService ->
+# {
+#   required_fields: {},
+#   invariants: [
+#       "persist_task creates or updates a record",
+#       "load_tasks retrieves valid Task instances",
+#       "delete_tasks correctly removes records from the database"
+#   ]
+# }
+# @TEST_FIXTURE: valid_task_persistence -> {"task_id": "123", "status": "PENDING"}
+# @TEST_EDGE: persist_invalid_task_type -> raises Exception
+# @TEST_EDGE: load_corrupt_json_params -> handled gracefully
+# @TEST_INVARIANT: accurate_round_trip -> verifies: [valid_task_persistence, load_corrupt_json_params]
 class TaskPersistenceService:
     # [DEF:_json_load_if_needed:Function]
     # @PURPOSE: Safely load JSON strings from DB if necessary
@@ -245,6 +259,19 @@ class TaskPersistenceService:
 # @TIER: CRITICAL
 # @RELATION: DEPENDS_ON -> TaskLogRecord
 # @INVARIANT: Log entries are batch-inserted for performance.
+#
+# @TEST_CONTRACT: TaskLogPersistenceService ->
+# {
+#   required_fields: {},
+#   invariants: [
+#       "add_logs efficiently saves logs to the database",
+#       "get_logs retrieves properly filtered LogEntry objects"
+#   ]
+# }
+# @TEST_FIXTURE: valid_log_batch -> {"task_id": "123", "logs": [{"level": "INFO", "message": "msg"}]}
+# @TEST_EDGE: empty_log_list -> no-op behavior
+# @TEST_EDGE: add_logs_db_error -> rollback and log error
+# @TEST_INVARIANT: accurate_log_aggregation -> verifies: [valid_log_batch]
 class TaskLogPersistenceService:
     """
     Service for persisting and querying task logs.

backend/src/core/task_manager/task_logger.py

@@ -15,8 +15,21 @@ from typing import Dict, Any, Optional, Callable
 # @PURPOSE: A wrapper around TaskManager._add_log that carries task_id and source context.
 # @TIER: CRITICAL
 # @INVARIANT: All log calls include the task_id and source.
-# @TEST_DATA: task_logger -> {"task_id": "test_123", "source": "test_plugin"}
 # @UX_STATE: Idle -> Logging -> (system records log)
+#
+# @TEST_CONTRACT: TaskLoggerModel ->
+# {
+#   required_fields: {task_id: str, add_log_fn: Callable},
+#   optional_fields: {source: str},
+#   invariants: [
+#       "All specific log methods (info, error) delegate to _log",
+#       "with_source creates a new logger with the same task_id"
+#   ]
+# }
+# @TEST_FIXTURE: valid_task_logger -> {"task_id": "test_123", "add_log_fn": lambda *args: None, "source": "test_plugin"}
+# @TEST_EDGE: missing_task_id -> raises TypeError
+# @TEST_EDGE: invalid_add_log_fn -> raises TypeError
+# @TEST_INVARIANT: consistent_delegation -> verifies: [valid_task_logger]
 class TaskLogger:
     """
     A dedicated logger for tasks that automatically tags logs with source attribution.

backend/src/models/report.py

@@ -47,6 +47,19 @@ class ReportStatus(str, Enum):
 # @INVARIANT: The properties accurately describe error state.
 # @SEMANTICS: error, context, payload
 # @PURPOSE: Error and recovery context for failed/partial reports.
+#
+# @TEST_CONTRACT: ErrorContextModel ->
+# {
+#   required_fields: {
+#       message: str
+#   },
+#   optional_fields: {
+#       code: str,
+#       next_actions: list[str]
+#   }
+# }
+# @TEST_FIXTURE: basic_error -> {"message": "Connection timeout", "code": "ERR_504", "next_actions": ["retry"]}
+# @TEST_EDGE: missing_message -> {"code": "ERR_504"}
 class ErrorContext(BaseModel):
     code: Optional[str] = None
     message: str
@@ -59,6 +72,36 @@ class ErrorContext(BaseModel):
 # @INVARIANT: Must represent canonical task record attributes.
 # @SEMANTICS: report, model, summary
 # @PURPOSE: Canonical normalized report envelope for one task execution.
+#
+# @TEST_CONTRACT: TaskReportModel ->
+# {
+#   required_fields: {
+#       report_id: str,
+#       task_id: str,
+#       task_type: TaskType,
+#       status: ReportStatus,
+#       updated_at: datetime,
+#       summary: str
+#   },
+#   invariants: [
+#       "report_id is a non-empty string",
+#       "task_id is a non-empty string",
+#       "summary is a non-empty string"
+#   ]
+# }
+# @TEST_FIXTURE: valid_task_report ->
+# {
+#   report_id: "rep-123",
+#   task_id: "task-456",
+#   task_type: "migration",
+#   status: "success",
+#   updated_at: "2026-02-26T12:00:00Z",
+#   summary: "Migration completed successfully"
+# }
+# @TEST_EDGE: empty_report_id -> {"report_id": "  ", "task_id": "task-456", "task_type": "migration", "status": "success", "updated_at": "2026-02-26T12:00:00Z", "summary": "Done"}
+# @TEST_EDGE: empty_summary -> {"report_id": "rep-123", "task_id": "task-456", "task_type": "migration", "status": "success", "updated_at": "2026-02-26T12:00:00Z", "summary": ""}
+# @TEST_EDGE: invalid_task_type -> {"report_id": "rep-123", "task_id": "task-456", "task_type": "invalid_type", "status": "success", "updated_at": "2026-02-26T12:00:00Z", "summary": "Done"}
+# @TEST_INVARIANT: non_empty_validators -> verifies: [empty_report_id, empty_summary]
 class TaskReport(BaseModel):
     report_id: str
     task_id: str
@@ -85,6 +128,25 @@ class TaskReport(BaseModel):
 # @INVARIANT: Time and pagination queries are mutually consistent.
 # @SEMANTICS: query, filter, search
 # @PURPOSE: Query object for server-side report filtering, sorting, and pagination.
+#
+# @TEST_CONTRACT: ReportQueryModel ->
+# {
+#   optional_fields: {
+#       page: int, page_size: int, task_types: list[TaskType], statuses: list[ReportStatus],
+#       time_from: datetime, time_to: datetime, search: str, sort_by: str, sort_order: str
+#   },
+#   invariants: [
+#       "page >= 1", "1 <= page_size <= 100",
+#       "sort_by in {'updated_at', 'status', 'task_type'}",
+#       "sort_order in {'asc', 'desc'}",
+#       "time_from <= time_to if both exist"
+#   ]
+# }
+# @TEST_FIXTURE: valid_query -> {"page": 1, "page_size":20, "sort_by": "updated_at", "sort_order": "desc"}
+# @TEST_EDGE: invalid_page_size_large -> {"page_size": 150}
+# @TEST_EDGE: invalid_sort_by -> {"sort_by": "unknown_field"}
+# @TEST_EDGE: invalid_time_range -> {"time_from": "2026-02-26T12:00:00Z", "time_to": "2026-02-25T12:00:00Z"}
+# @TEST_INVARIANT: attribute_constraints_enforced -> verifies: [invalid_page_size_large, invalid_sort_by, invalid_time_range]
 class ReportQuery(BaseModel):
     page: int = Field(default=1, ge=1)
     page_size: int = Field(default=20, ge=1, le=100)
@@ -124,6 +186,16 @@ class ReportQuery(BaseModel):
 # @INVARIANT: Represents paginated data correctly.
 # @SEMANTICS: collection, pagination
 # @PURPOSE: Paginated collection of normalized task reports.
+#
+# @TEST_CONTRACT: ReportCollectionModel ->
+# {
+#   required_fields: {
+#       items: list[TaskReport], total: int, page: int, page_size: int, has_next: bool, applied_filters: ReportQuery
+#   },
+#   invariants: ["total >= 0", "page >= 1", "page_size >= 1"]
+# }
+# @TEST_FIXTURE: empty_collection -> {"items": [], "total": 0, "page": 1, "page_size": 20, "has_next": False, "applied_filters": {}}
+# @TEST_EDGE: negative_total -> {"items": [], "total": -5, "page": 1, "page_size": 20, "has_next": False, "applied_filters": {}}
 class ReportCollection(BaseModel):
     items: List[TaskReport]
     total: int = Field(ge=0)
@@ -139,6 +211,14 @@ class ReportCollection(BaseModel):
 # @INVARIANT: Incorporates a report and logs correctly.
 # @SEMANTICS: view, detail, logs
 # @PURPOSE: Detailed report representation including diagnostics and recovery actions.
+#
+# @TEST_CONTRACT: ReportDetailViewModel ->
+# {
+#   required_fields: {report: TaskReport},
+#   optional_fields: {timeline: list[dict], diagnostics: dict, next_actions: list[str]}
+# }
+# @TEST_FIXTURE: valid_detail -> {"report": {"report_id": "rep-1", "task_id": "task-1", "task_type": "backup", "status": "success", "updated_at": "2026-02-26T12:00:00Z", "summary": "Done"}}
+# @TEST_EDGE: missing_report -> {}
 class ReportDetailView(BaseModel):
     report: TaskReport
     timeline: List[Dict[str, Any]] = Field(default_factory=list)

backend/src/models/task.py

@@ -39,6 +39,61 @@ class TaskRecord(Base):
 # @TIER: CRITICAL
 # @RELATION: DEPENDS_ON -> TaskRecord
 # @INVARIANT: Each log entry belongs to exactly one task.
+#
+# @TEST_CONTRACT: TaskLogCreate ->
+# {
+#   required_fields: {
+#       task_id: str,
+#       timestamp: datetime,
+#       level: str,
+#       source: str,
+#       message: str
+#   },
+#   optional_fields: {
+#       metadata_json: str,
+#       id: int
+#   },
+#   invariants: [
+#       "task_id matches an existing TaskRecord.id"
+#   ]
+# }
+#
+# @TEST_FIXTURE: basic_info_log ->
+# {
+#   task_id: "00000000-0000-0000-0000-000000000000",
+#   timestamp: "2026-02-26T12:00:00Z",
+#   level: "INFO",
+#   source: "system",
+#   message: "Task initialization complete"
+# }
+#
+# @TEST_EDGE: missing_required_field ->
+# {
+#   timestamp: "2026-02-26T12:00:00Z",
+#   level: "ERROR",
+#   source: "system",
+#   message: "Missing task_id"
+# }
+#
+# @TEST_EDGE: invalid_type ->
+# {
+#   task_id: "00000000-0000-0000-0000-000000000000",
+#   timestamp: "2026-02-26T12:00:00Z",
+#   level: 500,
+#   source: "system",
+#   message: "Integer level"
+# }
+#
+# @TEST_EDGE: empty_message ->
+# {
+#   task_id: "00000000-0000-0000-0000-000000000000",
+#   timestamp: "2026-02-26T12:00:00Z",
+#   level: "DEBUG",
+#   source: "system",
+#   message: ""
+# }
+#
+# @TEST_INVARIANT: exact_one_task_association -> verifies: [basic_info_log, missing_required_field]
 class TaskLogRecord(Base):
     __tablename__ = "task_logs"
 

backend/src/services/llm_provider.py

@@ -18,6 +18,18 @@ import os
 # @TIER: CRITICAL
 # @PURPOSE: Handles encryption and decryption of sensitive data like API keys.
 # @INVARIANT: Uses a secret key from environment or a default one (fallback only for dev).
+#
+# @TEST_CONTRACT: EncryptionManagerModel ->
+# {
+#   required_fields: {},
+#   invariants: [
+#       "encrypted data can be decrypted back to the original string"
+#   ]
+# }
+# @TEST_FIXTURE: basic_encryption_cycle -> {"data": "my_secret_key"}
+# @TEST_EDGE: decrypt_invalid_data -> raises Exception
+# @TEST_EDGE: empty_string_encryption -> {"data": ""}
+# @TEST_INVARIANT: symmetric_encryption -> verifies: [basic_encryption_cycle, empty_string_encryption]
 class EncryptionManager:
     # [DEF:EncryptionManager.__init__:Function]
     # @PURPOSE: Initialize the encryption manager with a Fernet key.

backend/src/services/reports/normalizer.py

@@ -113,6 +113,20 @@ def extract_error_context(task: Task, report_status: ReportStatus) -> Optional[E
 # @POST: Returns TaskReport with required fields and deterministic fallback behavior.
 # @PARAM: task (Task) - Source task.
 # @RETURN: TaskReport - Canonical normalized report.
+#
+# @TEST_CONTRACT: NormalizeTaskReport ->
+# {
+#   required_fields: {task: Task},
+#   invariants: [
+#       "Returns a valid TaskReport object",
+#       "Maps TaskStatus to ReportStatus deterministically",
+#       "Extracts ErrorContext for FAILED/PARTIAL tasks"
+#   ]
+# }
+# @TEST_FIXTURE: valid_task -> {"task": "MockTask(id='1', plugin_id='superset-migration', status=TaskStatus.SUCCESS)"}
+# @TEST_EDGE: task_with_error -> {"task": "MockTask(status=TaskStatus.FAILED, logs=[LogEntry(level='ERROR', message='Failed')])"}
+# @TEST_EDGE: unknown_plugin_type -> {"task": "MockTask(plugin_id='unknown-plugin', status=TaskStatus.PENDING)"}
+# @TEST_INVARIANT: deterministic_normalization -> verifies: [valid_task, task_with_error, unknown_plugin_type]
 def normalize_task_report(task: Task) -> TaskReport:
     with belief_scope("normalize_task_report"):
         task_type = resolve_task_type(task.plugin_id)

backend/src/services/reports/report_service.py

@@ -26,6 +26,19 @@ from .normalizer import normalize_task_report
 # @PRE: TaskManager dependency is initialized.
 # @POST: Provides deterministic list/detail report responses.
 # @INVARIANT: Service methods are read-only over task history source.
+#
+# @TEST_CONTRACT: ReportsServiceModel ->
+# {
+#   required_fields: {task_manager: TaskManager},
+#   invariants: [
+#       "list_reports returns a matching ReportCollection",
+#       "get_report_detail returns a valid ReportDetailView or None"
+#   ]
+# }
+# @TEST_FIXTURE: valid_service -> {"task_manager": "MockTaskManager()"}
+# @TEST_EDGE: empty_task_list -> returns empty ReportCollection
+# @TEST_EDGE: report_not_found -> get_report_detail returns None
+# @TEST_INVARIANT: consistent_pagination -> verifies: [valid_service]
 class ReportsService:
     # [DEF:__init__:Function]
     # @TIER: CRITICAL

backend/src/services/reports/type_profiles.py

@@ -71,6 +71,17 @@ TASK_TYPE_PROFILES: Dict[TaskType, Dict[str, Any]] = {
 # @POST: Always returns one of TaskType enum values.
 # @PARAM: plugin_id (Optional[str]) - Source plugin/task identifier from task record.
 # @RETURN: TaskType - Resolved canonical type or UNKNOWN fallback.
+#
+# @TEST_CONTRACT: ResolveTaskType ->
+# {
+#   required_fields: {plugin_id: str},
+#   invariants: ["returns TaskType.UNKNOWN for missing/unmapped plugin_id"]
+# }
+# @TEST_FIXTURE: valid_plugin -> {"plugin_id": "superset-migration"}
+# @TEST_EDGE: empty_plugin -> {"plugin_id": ""}
+# @TEST_EDGE: none_plugin -> {"plugin_id": None}
+# @TEST_EDGE: unknown_plugin -> {"plugin_id": "invalid-plugin"}
+# @TEST_INVARIANT: fallback_to_unknown -> verifies: [empty_plugin, none_plugin, unknown_plugin]
 def resolve_task_type(plugin_id: Optional[str]) -> TaskType:
     with belief_scope("resolve_task_type"):
         normalized = (plugin_id or "").strip()
@@ -86,6 +97,15 @@ def resolve_task_type(plugin_id: Optional[str]) -> TaskType:
 # @POST: Returns a profile dict and never raises for unknown types.
 # @PARAM: task_type (TaskType) - Canonical task type.
 # @RETURN: Dict[str, Any] - Profile metadata used by normalization and UI contracts.
+#
+# @TEST_CONTRACT: GetTypeProfile ->
+# {
+#   required_fields: {task_type: TaskType},
+#   invariants: ["returns a valid metadata dictionary even for UNKNOWN"]
+# }
+# @TEST_FIXTURE: valid_profile -> {"task_type": "migration"}
+# @TEST_EDGE: missing_profile -> {"task_type": "some_new_type"}
+# @TEST_INVARIANT: always_returns_dict -> verifies: [valid_profile, missing_profile]
 def get_type_profile(task_type: TaskType) -> Dict[str, Any]:
     with belief_scope("get_type_profile"):
         return TASK_TYPE_PROFILES.get(task_type, TASK_TYPE_PROFILES[TaskType.UNKNOWN])