busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

logs: migrate to molecular CoT JSON protocol, fix regressions

Browse Source 
- Replace BeliefFormatter with CotJsonFormatter (single-line JSON with
  ts, level, trace_id, src, marker, intent, payload, error, span_id)
- Migrate belief_scope to use cot_log() for structured JSON output
- Update monkey-patched reason/reflect/explore to pass extra= dict
- Strip 200+ inline [REASON]/[REFLECT]/[EXPLORE] markers from 50+ files
- Remove belief_scope from hot-path utilities (_parse_datetime,
  _json_load_if_needed) to eliminate startup log noise
- Register TraceContextMiddleware in app.py (was implemented but unused)
- Seed trace_id in startup_event/shutdown_event for background tasks
- Fix broken relative imports in translate routes (3 dots → 4 dots)
- Migrate 43 translate_routes log calls to logger.reason/explore
- Fix SyntaxError (positional arg after extra=) in _repo_operations_routes
- Fix IndentationError in rbac_permission_catalog except-block
- Add smoke test tests/test_smoke_app.py (catches import errors before run)
This commit is contained in:
busya
2026-05-13 09:44:50 +03:00
parent b6f46fe9f5
commit 83b8f4d999
55 changed files with 644 additions and 539 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
backend/src/api/routes/admin.py
View File

@@ -310,9 +310,9 @@ async def list_permissions(
db=db, declared_permissions=declared_permissions
)
if inserted_count > 0:
logger.info(
"[api.admin.list_permissions][REASON] Synchronized %s missing RBAC permissions into auth catalog",
inserted_count,
logger.reason(
f"Synchronized {inserted_count} missing RBAC permissions into auth catalog",
extra={"src": "api.admin.list_permissions"},
)
repo = AuthRepository(db)

26
backend/src/api/routes/dashboards/_listing_routes.py
View File

@@ -141,7 +141,8 @@ async def get_dashboards(
)
except Exception as profile_error:
logger.explore(
f"[EXPLORE] Profile preference unavailable; continuing without profile-default filter: {profile_error}"
f"Profile preference unavailable; continuing without profile-default filter: {profile_error}",
extra={"src": "get_dashboards"},
)
try:
@@ -183,9 +184,9 @@ async def get_dashboards(
total = page_payload["total"]
total_pages = page_payload["total_pages"]
except Exception as page_error:
logger.warning(
"[get_dashboards][REASON] Page-based fetch failed; using compatibility fallback: %s",
page_error,
logger.reason(
f"Page-based fetch failed; using compatibility fallback: {page_error}",
extra={"src": "get_dashboards"},
)
if can_apply_slug_filter:
dashboards = await resource_service.get_dashboards_with_status(
@@ -241,9 +242,8 @@ async def get_dashboards(
if not actor_aliases:
actor_aliases = [bound_username]
logger.reason(
"[REASON] Applying profile actor filter "
f"(env={env_id}, bound_username={bound_username}, actor_aliases={actor_aliases!r}, "
f"dashboards_before={len(dashboards)})"
"Applying profile actor filter",
extra={"src": "get_dashboards", "payload": {"env": env_id, "bound_username": bound_username, "actor_aliases": actor_aliases, "dashboards_before": len(dashboards)}},
)
filtered_dashboards: List[Dict[str, Any]] = []
max_actor_samples = 15
@@ -259,19 +259,15 @@ async def get_dashboards(
)
if index < max_actor_samples:
logger.reflect(
"[REFLECT] Profile actor filter sample "
f"(env={env_id}, dashboard_id={dashboard.get('id')}, "
f"bound_username={bound_username!r}, actor_aliases={actor_aliases!r}, "
f"owners={owners_value!r}, created_by={created_by_value!r}, "
f"modified_by={modified_by_value!r}, matches={matches_actor})"
"Profile actor filter sample",
extra={"src": "get_dashboards", "payload": {"env": env_id, "dashboard_id": dashboard.get('id'), "bound_username": bound_username, "actor_aliases": actor_aliases, "owners": owners_value, "created_by": created_by_value, "modified_by": modified_by_value, "matches": matches_actor}},
)
if matches_actor:
filtered_dashboards.append(dashboard)
logger.reflect(
"[REFLECT] Profile actor filter summary "
f"(env={env_id}, bound_username={bound_username!r}, "
f"dashboards_before={len(dashboards)}, dashboards_after={len(filtered_dashboards)})"
"Profile actor filter summary",
extra={"src": "get_dashboards", "payload": {"env": env_id, "bound_username": bound_username, "dashboards_before": len(dashboards), "dashboards_after": len(filtered_dashboards)}},
)
dashboards = filtered_dashboards

9
backend/src/api/routes/dashboards/_projection.py
View File

@@ -182,14 +182,13 @@ def _resolve_profile_actor_aliases(env: Any, bound_username: str) -> List[str]:
if display_alias and display_alias not in aliases:
aliases.append(display_alias)
logger.reflect(
"[REFLECT] Resolved profile actor aliases "
f"(env={getattr(env, 'id', None)}, bound_username={normalized_bound!r}, "
f"lookup_items={len(lookup_items)}, aliases={aliases!r})"
"Resolved profile actor aliases",
extra={"src": "_resolve_profile_actor_aliases", "payload": {"env": getattr(env, 'id', None), "bound_username": normalized_bound, "lookup_items": len(lookup_items), "aliases": aliases}},
)
except Exception as alias_error:
logger.explore(
"[EXPLORE] Failed to resolve profile actor aliases via Superset users lookup "
f"(env={getattr(env, 'id', None)}, bound_username={normalized_bound!r}): {alias_error}"
"Failed to resolve profile actor aliases via Superset users lookup",
extra={"src": "_resolve_profile_actor_aliases", "payload": {"env": getattr(env, 'id', None), "bound_username": normalized_bound}, "error": str(alias_error)},
)
return aliases

17
backend/src/api/routes/git/_helpers.py
View File

@@ -61,8 +61,9 @@ def _resolve_repository_status(dashboard_id: int) -> dict:
git_service = get_git_service()
repo_path = git_service._get_repo_path(dashboard_id)
if not os.path.exists(repo_path):
logger.debug(
f"[get_repository_status][REASON] Repository is not initialized for dashboard {dashboard_id}"
logger.reason(
f"Repository is not initialized for dashboard {dashboard_id}",
extra={"src": "get_repository_status"},
)
return _build_no_repo_status_payload()
@@ -70,8 +71,9 @@ def _resolve_repository_status(dashboard_id: int) -> dict:
return git_service.get_status(dashboard_id)
except HTTPException as e:
if e.status_code == 404:
logger.debug(
f"[get_repository_status][REASON] Repository is not initialized for dashboard {dashboard_id}"
logger.reason(
f"Repository is not initialized for dashboard {dashboard_id}",
extra={"src": "get_repository_status"},
)
return _build_no_repo_status_payload()
raise
@@ -287,10 +289,9 @@ def _resolve_current_user_git_identity(
.first()
)
except Exception as resolve_error:
logger.warning(
"[_resolve_current_user_git_identity][REASON] Failed to load profile preference for user %s: %s",
user_id,
resolve_error,
logger.explore(
"Failed to load profile preference for resolving git identity",
extra={"src": "_resolve_current_user_git_identity", "payload": {"user_id": user_id}, "error": str(resolve_error)},
)
return None

33
backend/src/api/routes/git/_repo_operations_routes.py
View File

@@ -128,23 +128,17 @@ async def pull_changes(
config_url = config_row.url
config_provider = config_row.provider
except Exception as diagnostics_error:
logger.warning(
"[pull_changes][REASON] Failed to load repository binding diagnostics for dashboard %s: %s",
dashboard_id,
diagnostics_error,
logger.explore(
"Failed to load repository binding diagnostics",
extra={"src": "pull_changes", "payload": {"dashboard_id": dashboard_id}, "error": str(diagnostics_error)},
)
logger.info(
"[pull_changes][REASON] Route diagnostics dashboard_ref=%s env_id=%s resolved_dashboard_id=%s "
"binding_exists=%s binding_local_path=%s binding_remote_url=%s binding_config_id=%s config_provider=%s config_url=%s",
dashboard_ref,
env_id,
dashboard_id,
bool(db_repo),
(db_repo.local_path if db_repo else None),
(db_repo.remote_url if db_repo else None),
(db_repo.config_id if db_repo else None),
config_provider,
config_url,
logger.reason(
f"Route diagnostics dashboard_ref={dashboard_ref} env_id={env_id} resolved_dashboard_id={dashboard_id} "
f"binding_exists={bool(db_repo)} binding_local_path={(db_repo.local_path if db_repo else None)} "
f"binding_remote_url={(db_repo.remote_url if db_repo else None)} "
f"binding_config_id={(db_repo.config_id if db_repo else None)} "
f"config_provider={config_provider} config_url={config_url}",
extra={"src": "pull_changes"},
)
_apply_git_identity_from_profile(dashboard_id, db, current_user)
_gs.pull_changes(dashboard_id)
@@ -190,10 +184,9 @@ async def get_repository_status_batch(
with belief_scope("get_repository_status_batch"):
dashboard_ids = list(dict.fromkeys(request.dashboard_ids))
if len(dashboard_ids) > MAX_REPOSITORY_STATUS_BATCH:
logger.warning(
"[get_repository_status_batch][REASON] Batch size %s exceeds limit %s. Truncating request.",
len(dashboard_ids),
MAX_REPOSITORY_STATUS_BATCH,
logger.reason(
f"Batch size {len(dashboard_ids)} exceeds limit {MAX_REPOSITORY_STATUS_BATCH}. Truncating request.",
extra={"src": "get_repository_status_batch"},
)
dashboard_ids = dashboard_ids[:MAX_REPOSITORY_STATUS_BATCH]

5
backend/src/api/routes/git/_repo_routes.py
View File

@@ -61,8 +61,9 @@ async def init_repository(
raise HTTPException(status_code=404, detail="Git configuration not found")
try:
logger.info(
f"[init_repository][REASON] Initializing repo for dashboard {dashboard_id}"
logger.reason(
f"Initializing repo for dashboard {dashboard_id}",
extra={"src": "init_repository"},
)
_gs.init_repo(
dashboard_id,

15
backend/src/api/routes/llm.py
View File

@@ -52,8 +52,9 @@ async def get_providers(
"""
Get all LLM provider configurations.
"""
logger.info(
f"[llm_routes][get_providers][REASON] Fetching providers for user: {current_user.username}"
logger.reason(
f"Fetching providers for user: {current_user.username}",
extra={"src": "llm_routes.get_providers"},
)
service = LLMProviderService(db)
providers = service.get_all_providers()
@@ -248,8 +249,9 @@ async def test_connection(
current_user: User = Depends(get_current_active_user),
db: Session = Depends(get_db),
):
logger.info(
f"[llm_routes][test_connection][REASON] Testing connection for provider_id: {provider_id}"
logger.reason(
f"Testing connection for provider_id: {provider_id}",
extra={"src": "llm_routes.test_connection"},
)
"""
Test connection to an LLM provider.
@@ -305,8 +307,9 @@ async def test_provider_config(
"""
from ...plugins.llm_analysis.service import LLMClient
logger.info(
f"[llm_routes][test_provider_config][REASON] Testing config for {config.name}"
logger.reason(
f"Testing config for {config.name}",
extra={"src": "llm_routes.test_provider_config"},
)
# Check if API key is provided

7
backend/src/api/routes/migration.py
View File

@@ -357,8 +357,9 @@ async def trigger_sync_now(
credentials_id=env.id, # Use env.id as credentials reference
)
db.add(db_env)
logger.info(
f"[trigger_sync_now][REASON] Created environment row for {env.id}"
logger.reason(
f"Created environment row for {env.id}",
extra={"src": "trigger_sync_now"},
)
else:
existing.name = env.name
@@ -373,7 +374,7 @@ async def trigger_sync_now(
client = SupersetClient(env)
service.sync_environment(env.id, client)
results["synced"].append(env.id)
logger.info(f"[trigger_sync_now][REASON] Synced environment {env.id}")
logger.reason(f"Synced environment {env.id}", extra={"src": "trigger_sync_now"})
except Exception as e:
results["failed"].append({"env_id": env.id, "error": str(e)})
logger.error(f"[trigger_sync_now][Error] Failed to sync {env.id}: {e}")

12
backend/src/api/routes/profile.py
View File

@@ -69,7 +69,7 @@ async def get_preferences(
plugin_loader=Depends(get_plugin_loader),
):
with belief_scope("profile.get_preferences", f"user_id={current_user.id}"):
logger.reason("[REASON] Resolving current user preference")
logger.reason("Resolving current user preference")
service = _get_profile_service(db, config_manager, plugin_loader)
return service.get_my_preference(current_user)
# #endregion get_preferences
@@ -91,13 +91,13 @@ async def update_preferences(
with belief_scope("profile.update_preferences", f"user_id={current_user.id}"):
service = _get_profile_service(db, config_manager, plugin_loader)
try:
logger.reason("[REASON] Attempting preference save")
logger.reason("Attempting preference save")
return service.update_my_preference(current_user=current_user, payload=payload)
except ProfileValidationError as exc:
logger.reflect("[REFLECT] Preference validation failed")
logger.reflect("Preference validation failed")
raise HTTPException(status_code=422, detail=exc.errors) from exc
except ProfileAuthorizationError as exc:
logger.explore("[EXPLORE] Cross-user mutation guard blocked request")
logger.explore("Cross-user mutation guard blocked request")
raise HTTPException(status_code=403, detail=str(exc)) from exc
# #endregion update_preferences
@@ -134,13 +134,13 @@ async def lookup_superset_accounts(
sort_order=sort_order,
)
try:
logger.reason("[REASON] Executing Superset account lookup")
logger.reason("Executing Superset account lookup")
return service.lookup_superset_accounts(
current_user=current_user,
request=lookup_request,
)
except EnvironmentNotFoundError as exc:
logger.explore("[EXPLORE] Lookup request references unknown environment")
logger.explore("Lookup request references unknown environment")
raise HTTPException(status_code=404, detail=str(exc)) from exc
# #endregion lookup_superset_accounts

22
backend/src/api/routes/settings.py
View File

@@ -88,7 +88,7 @@ async def get_settings(
_=Depends(has_permission("admin:settings", "READ")),
):
with belief_scope("get_settings"):
logger.info("[get_settings][REASON] Fetching all settings")
logger.reason("Fetching all settings", extra={"src": "get_settings"})
config = config_manager.get_config().copy(deep=True)
config.settings.llm = normalize_llm_settings(config.settings.llm)
# Mask passwords
@@ -127,7 +127,7 @@ async def update_global_settings(
_=Depends(has_permission("admin:settings", "WRITE")),
):
with belief_scope("update_global_settings"):
logger.info("[update_global_settings][REASON] Updating global settings")
logger.reason("Updating global settings", extra={"src": "update_global_settings"})
config_manager.update_global_settings(settings)
return settings
@@ -183,7 +183,7 @@ async def get_environments(
_=Depends(has_permission("admin:settings", "READ")),
):
with belief_scope("get_environments"):
logger.info("[get_environments][REASON] Fetching environments")
logger.reason("Fetching environments", extra={"src": "get_environments"})
environments = config_manager.get_environments()
return [
env.copy(update={"url": _normalize_superset_env_url(env.url)})
@@ -205,7 +205,7 @@ async def add_environment(
_=Depends(has_permission("admin:settings", "WRITE")),
):
with belief_scope("add_environment"):
logger.info(f"[add_environment][REASON] Adding environment {env.id}")
logger.reason(f"Adding environment {env.id}", extra={"src": "add_environment"})
env = env.copy(update={"url": _normalize_superset_env_url(env.url)})
# Validate connection before adding (fast path)
@@ -237,7 +237,7 @@ async def update_environment(
config_manager: ConfigManager = Depends(get_config_manager),
):
with belief_scope("update_environment"):
logger.info(f"[update_environment][REASON] Updating environment {id}")
logger.reason(f"Updating environment {id}", extra={"src": "update_environment"})
env = env.copy(update={"url": _normalize_superset_env_url(env.url)})
@@ -278,7 +278,7 @@ async def delete_environment(
id: str, config_manager: ConfigManager = Depends(get_config_manager)
):
with belief_scope("delete_environment"):
logger.info(f"[delete_environment][REASON] Deleting environment {id}")
logger.reason(f"Deleting environment {id}", extra={"src": "delete_environment"})
config_manager.delete_environment(id)
return {"message": f"Environment {id} deleted"}
@@ -295,7 +295,7 @@ async def test_environment_connection(
id: str, config_manager: ConfigManager = Depends(get_config_manager)
):
with belief_scope("test_environment_connection"):
logger.info(f"[test_environment_connection][REASON] Testing environment {id}")
logger.reason(f"Testing environment {id}", extra={"src": "test_environment_connection"})
# Find environment
env = next((e for e in config_manager.get_environments() if e.id == id), None)
@@ -351,8 +351,8 @@ async def update_logging_config(
_=Depends(has_permission("admin:settings", "WRITE")),
):
with belief_scope("update_logging_config"):
logger.info(
f"[update_logging_config][REASON] Updating logging config: level={config.level}, task_log_level={config.task_log_level}"
logger.reason(
f"Updating logging config: level={config.level}, task_log_level={config.task_log_level}", extra={"src": "update_logging_config"}
)
# Get current settings and update logging config
@@ -475,8 +475,8 @@ async def update_consolidated_settings(
_=Depends(has_permission("admin:settings", "WRITE")),
):
with belief_scope("update_consolidated_settings"):
logger.info(
"[update_consolidated_settings][REASON] Applying consolidated settings patch"
logger.reason(
"Applying consolidated settings patch", extra={"src": "update_consolidated_settings"}
)
current_config = config_manager.get_config()

4
backend/src/api/routes/translate/_correction_routes.py
View File

@@ -36,7 +36,7 @@ async def submit_correction(
db: Session = Depends(get_db),
):
"""Submit a term correction from a run result review."""
logger.info(f"[translate_routes][submit_correction] User: {current_user.username}")
logger.reason(f"submit_correction — User: {current_user.username}", extra={"src": "translate_routes"})
if not payload.dictionary_id:
raise HTTPException(status_code=422, detail="dictionary_id is required")
try:
@@ -68,7 +68,7 @@ async def submit_bulk_corrections(
db: Session = Depends(get_db),
):
"""Submit multiple term corrections atomically."""
logger.info(f"[translate_routes][submit_bulk_corrections] User: {current_user.username}")
logger.reason(f"submit_bulk_corrections — User: {current_user.username}", extra={"src": "translate_routes"})
try:
corr_list = []
for c in payload.corrections:

4
backend/src/api/routes/translate/_dictionary_routes.py
View File

@@ -67,7 +67,7 @@ async def get_dictionary(
logger.reason(f"Dict: {dictionary_id}, User: {current_user.username}")
try:
d = DictionaryManager.get_dictionary(db, dictionary_id)
from ...models.translate import DictionaryEntry
from ....models.translate import DictionaryEntry
entry_count = db.query(DictionaryEntry.id).filter(DictionaryEntry.dictionary_id == dictionary_id).count()
logger.reflect("Dictionary found", {"id": dictionary_id})
return _dict_to_response(d, entry_count)
@@ -129,7 +129,7 @@ async def update_dictionary(
target_dialect=payload.target_dialect,
is_active=payload.is_active,
)
from ...models.translate import DictionaryEntry
from ....models.translate import DictionaryEntry
entry_count = db.query(DictionaryEntry.id).filter(DictionaryEntry.dictionary_id == dictionary_id).count()
logger.reflect("Dictionary updated", {"id": dictionary_id})
return _dict_to_response(d, entry_count)

2
backend/src/api/routes/translate/_helpers.py
View File

@@ -57,7 +57,7 @@ def _dict_to_response(d: Any, entry_count: int = 0) -> dict:
# @BRIEF Get entry counts for a list of dictionary IDs.
def _get_dictionary_entry_counts(db: Session, dict_ids: List[str]) -> Dict[str, int]:
from sqlalchemy import func
from ...models.translate import DictionaryEntry
from ....models.translate import DictionaryEntry
counts = (
db.query(DictionaryEntry.dictionary_id, func.count(DictionaryEntry.id))
.filter(DictionaryEntry.dictionary_id.in_(dict_ids))

22
backend/src/api/routes/translate/_job_routes.py
View File

@@ -46,7 +46,7 @@ async def list_jobs(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""List all translation jobs."""
logger.info(f"[translate_routes][list_jobs] User: {current_user.username}")
logger.reason(f"list_jobs — User: {current_user.username}", extra={"src": "translate_routes"})
try:
service = TranslateJobService(db, config_manager, current_user.username)
total, jobs = service.list_jobs(
@@ -77,7 +77,7 @@ async def get_job(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Get a translation job by ID."""
logger.info(f"[translate_routes][get_job] Job: {job_id}, User: {current_user.username}")
logger.reason(f"get_job — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
service = TranslateJobService(db, config_manager, current_user.username)
job = service.get_job(job_id)
@@ -102,7 +102,7 @@ async def create_job(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Create a new translation job."""
logger.info(f"[translate_routes][create_job] User: {current_user.username}")
logger.reason(f"create_job — User: {current_user.username}", extra={"src": "translate_routes"})
try:
service = TranslateJobService(db, config_manager, current_user.username)
job = service.create_job(payload)
@@ -128,7 +128,7 @@ async def update_job(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Update a translation job."""
logger.info(f"[translate_routes][update_job] Job: {job_id}, User: {current_user.username}")
logger.reason(f"update_job — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
service = TranslateJobService(db, config_manager, current_user.username)
job = service.update_job(job_id, payload)
@@ -152,7 +152,7 @@ async def delete_job(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Delete a translation job."""
logger.info(f"[translate_routes][delete_job] Job: {job_id}, User: {current_user.username}")
logger.reason(f"delete_job — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
service = TranslateJobService(db, config_manager, current_user.username)
service.delete_job(job_id)
@@ -174,7 +174,7 @@ async def duplicate_job(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Duplicate a translation job."""
logger.info(f"[translate_routes][duplicate_job] Job: {job_id}, User: {current_user.username}")
logger.reason(f"duplicate_job — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
service = TranslateJobService(db, config_manager, current_user.username)
new_job = service.duplicate_job(job_id)
@@ -212,7 +212,7 @@ async def list_datasources(
datasets = service.fetch_available_datasources(env_id, search)
return datasets
except Exception as e:
logger.error(f"[translate_routes][list_datasources] Error: {e}")
logger.explore(f"list_datasources failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=str(e))
@router.get("/datasources/{datasource_id}/columns", response_model=DatasourceColumnsResponse)
@@ -224,9 +224,9 @@ async def get_datasource_columns(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Get column metadata and database dialect for a Superset datasource."""
logger.info(
f"[translate_routes][get_datasource_columns] "
f"Datasource: {datasource_id}, Env: {env_id}, User: {current_user.username}"
logger.reason(
f"get_datasource_columns — Datasource: {datasource_id}, Env: {env_id}, User: {current_user.username}",
extra={"src": "translate_routes"}
)
try:
result = fetch_datasource_columns_service(
@@ -238,7 +238,7 @@ async def get_datasource_columns(
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
except Exception as e:
logger.error(f"[translate_routes][get_datasource_columns] Error: {e}")
logger.explore(f"get_datasource_columns failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(
status_code=status.HTTP_502_BAD_GATEWAY,
detail=f"Failed to fetch datasource columns from Superset: {e}",

4
backend/src/api/routes/translate/_metrics_routes.py
View File

@@ -31,7 +31,7 @@ async def get_metrics(
db: Session = Depends(get_db),
):
"""Get translation metrics."""
logger.info(f"[translate_routes][get_metrics] User: {current_user.username}")
logger.reason(f"get_metrics — User: {current_user.username}", extra={"src": "translate_routes"})
try:
metrics_svc = TranslationMetrics(db)
if job_id:
@@ -54,7 +54,7 @@ async def get_job_metrics(
db: Session = Depends(get_db),
):
"""Get aggregated metrics for a specific job."""
logger.info(f"[translate_routes][get_job_metrics] Job: {job_id}, User: {current_user.username}")
logger.reason(f"get_job_metrics — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
metrics_svc = TranslationMetrics(db)
return metrics_svc.get_job_metrics(job_id)

18
backend/src/api/routes/translate/_preview_routes.py
View File

@@ -41,7 +41,7 @@ async def preview_translation(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Preview a translation before applying it."""
logger.info(f"[translate_routes][preview_translation] Job: {job_id}, User: {current_user.username}")
logger.reason(f"preview_translation — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
if payload is None:
payload = PreviewRequest()
try:
@@ -56,7 +56,7 @@ async def preview_translation(
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
except Exception as e:
logger.error(f"[translate_routes][preview_translation] Error: {e}")
logger.explore(f"preview_translation failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Preview failed: {e}")
# #endregion preview_translation
@@ -76,7 +76,7 @@ async def update_preview_row(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Approve, edit, or reject a preview row."""
logger.info(f"[translate_routes][update_preview_row] Job: {job_id}, Row: {row_key}, Action: {payload.action}")
logger.reason(f"update_preview_row — Job: {job_id}, Row: {row_key}, Action: {payload.action}", extra={"src": "translate_routes"})
try:
preview_service = TranslationPreview(db, config_manager, current_user.username)
result = preview_service.update_preview_row(
@@ -105,7 +105,7 @@ async def accept_preview_session(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Accept a preview session, enabling full translation execution."""
logger.info(f"[translate_routes][accept_preview_session] Job: {job_id}, User: {current_user.username}")
logger.reason(f"accept_preview_session — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
preview_service = TranslationPreview(db, config_manager, current_user.username)
result = preview_service.accept_preview_session(job_id=job_id)
@@ -128,9 +128,9 @@ async def apply_preview(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Apply a preview session by session ID."""
logger.info(f"[translate_routes][apply_preview] Session: {session_id}, User: {current_user.username}")
logger.reason(f"apply_preview — Session: {session_id}, User: {current_user.username}", extra={"src": "translate_routes"})
# Find job_id from session
from ...models.translate import TranslationPreviewSession
from ....models.translate import TranslationPreviewSession
session = db.query(TranslationPreviewSession).filter(TranslationPreviewSession.id == session_id).first()
if not session:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Preview session '{session_id}' not found")
@@ -159,9 +159,9 @@ async def get_preview_records(
db: Session = Depends(get_db),
):
"""Get records for a preview session."""
logger.info(f"[translate_routes][get_preview_records] Session: {session_id}, User: {current_user.username}")
logger.reason(f"get_preview_records — Session: {session_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
from ...models.translate import TranslationPreviewSession, TranslationPreviewRecord
from ....models.translate import TranslationPreviewSession, TranslationPreviewRecord
session = db.query(TranslationPreviewSession).filter(TranslationPreviewSession.id == session_id).first()
if not session:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Preview session '{session_id}' not found")
@@ -186,7 +186,7 @@ async def get_preview_records(
except HTTPException:
raise
except Exception as e:
logger.error(f"[translate_routes][get_preview_records] Error: {e}")
logger.explore(f"get_preview_records failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
# #endregion get_preview_records

16
backend/src/api/routes/translate/_run_list_routes.py
View File

@@ -40,9 +40,9 @@ async def list_runs(
db: Session = Depends(get_db),
):
"""List all runs with cross-job filtering and pagination."""
logger.info(f"[translate_routes][list_runs] User: {current_user.username}")
logger.reason(f"list_runs — User: {current_user.username}", extra={"src": "translate_routes"})
try:
from ...models.translate import TranslationRun
from ....models.translate import TranslationRun
query = db.query(TranslationRun)
if job_id:
@@ -100,7 +100,7 @@ async def list_runs(
return {"items": items, "total": total, "page": page, "page_size": page_size}
except Exception as e:
logger.error(f"[translate_routes][list_runs] Error: {e}")
logger.explore(f"list_runs failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
# #endregion list_runs
@@ -122,7 +122,7 @@ async def get_run_detail(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Get detailed run info with config_snapshot, records, events."""
logger.info(f"[translate_routes][get_run_detail] Run: {run_id}, User: {current_user.username}")
logger.reason(f"get_run_detail — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
status_info = orch.get_run_status(run_id)
@@ -131,7 +131,7 @@ async def get_run_detail(
event_summary = TranslationEventLog(db).get_run_event_summary(run_id)
# Get config snapshot from run
from ...models.translate import TranslationRun
from ....models.translate import TranslationRun
run = db.query(TranslationRun).filter(TranslationRun.id == run_id).first()
config_snapshot = run.config_snapshot if run else None
@@ -164,9 +164,9 @@ async def download_skipped_csv(
db: Session = Depends(get_db),
):
"""Download skipped translation records as CSV."""
logger.info(f"[translate_routes][download_skipped_csv] Run: {run_id}, User: {current_user.username}")
logger.reason(f"download_skipped_csv — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
from ...models.translate import TranslationRecord
from ....models.translate import TranslationRecord
from fastapi.responses import StreamingResponse
import csv
import io
@@ -197,7 +197,7 @@ async def download_skipped_csv(
headers={"Content-Disposition": f"attachment; filename=skipped_{run_id}.csv"},
)
except Exception as e:
logger.error(f"[translate_routes][download_skipped_csv] Error: {e}")
logger.explore(f"download_skipped_csv failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
# #endregion download_skipped_csv

26
backend/src/api/routes/translate/_run_routes.py
View File

@@ -35,7 +35,7 @@ async def run_translation(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Execute a translation job (trigger a run)."""
logger.info(f"[translate_routes][run_translation] Job: {job_id}, User: {current_user.username}")
logger.reason(f"run_translation — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
run = orch.start_run(job_id=job_id, is_scheduled=False)
@@ -50,7 +50,7 @@ async def run_translation(
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
except Exception as e:
logger.error(f"[translate_routes][run_translation] Error: {e}")
logger.explore(f"run_translation failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Run failed: {e}")
# #endregion run_translation
@@ -68,7 +68,7 @@ async def retry_run(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Retry failed batches in a translation run."""
logger.info(f"[translate_routes][retry_run] Run: {run_id}, User: {current_user.username}")
logger.reason(f"retry_run — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
run = orch.retry_failed_batches(run_id)
@@ -76,7 +76,7 @@ async def retry_run(
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
except Exception as e:
logger.error(f"[translate_routes][retry_run] Error: {e}")
logger.explore(f"retry_run failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Retry failed: {e}")
# #endregion retry_run
@@ -94,7 +94,7 @@ async def retry_insert(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Retry the SQL insert phase for a completed run."""
logger.info(f"[translate_routes][retry_insert] Run: {run_id}, User: {current_user.username}")
logger.reason(f"retry_insert — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
run = orch.retry_insert(run_id)
@@ -102,7 +102,7 @@ async def retry_insert(
except ValueError as e:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
except Exception as e:
logger.error(f"[translate_routes][retry_insert] Error: {e}")
logger.explore(f"retry_insert failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_502_BAD_GATEWAY, detail=f"Retry insert failed: {e}")
# #endregion retry_insert
@@ -120,7 +120,7 @@ async def cancel_run(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Cancel a running translation."""
logger.info(f"[translate_routes][cancel_run] Run: {run_id}, User: {current_user.username}")
logger.reason(f"cancel_run — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
run = orch.cancel_run(run_id)
@@ -145,7 +145,7 @@ async def get_run_history(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Get run history for a translation job."""
logger.info(f"[translate_routes][get_run_history] Job: {job_id}, User: {current_user.username}")
logger.reason(f"get_run_history — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
total, runs = orch.get_run_history(job_id, page=page, page_size=page_size)
@@ -168,7 +168,7 @@ async def get_run_status(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Get status and statistics for a translation run."""
logger.info(f"[translate_routes][get_run_status] Run: {run_id}, User: {current_user.username}")
logger.reason(f"get_run_status — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
return orch.get_run_status(run_id)
@@ -193,7 +193,7 @@ async def get_run_records(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Get paginated records for a translation run."""
logger.info(f"[translate_routes][get_run_records] Run: {run_id}, User: {current_user.username}")
logger.reason(f"get_run_records — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
orch = TranslationOrchestrator(db, config_manager, current_user.username)
return orch.get_run_records(run_id, page=page, page_size=page_size, status_filter=status)
@@ -218,9 +218,9 @@ async def get_batches(
db: Session = Depends(get_db),
):
"""Get batches for a translation run."""
logger.info(f"[translate_routes][get_batches] Run: {run_id}, User: {current_user.username}")
logger.reason(f"get_batches — Run: {run_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
from ...models.translate import TranslationBatch
from ....models.translate import TranslationBatch
batches = (
db.query(TranslationBatch)
.filter(TranslationBatch.run_id == run_id)
@@ -243,7 +243,7 @@ async def get_batches(
for b in batches
]
except Exception as e:
logger.error(f"[translate_routes][get_batches] Error: {e}")
logger.explore(f"get_batches failed", extra={"src": "translate_routes", "error": str(e)})
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
# #endregion get_batches

14
backend/src/api/routes/translate/_schedule_routes.py
View File

@@ -34,7 +34,7 @@ async def get_schedule(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Get schedule for a translation job."""
logger.info(f"[translate_routes][get_schedule] Job: {job_id}, User: {current_user.username}")
logger.reason(f"get_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
scheduler = TranslationScheduler(db, config_manager, current_user.username)
schedule = scheduler.get_schedule(job_id)
@@ -69,11 +69,11 @@ async def set_schedule(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Set or update schedule for a translation job."""
logger.info(f"[translate_routes][set_schedule] Job: {job_id}, User: {current_user.username}")
logger.reason(f"set_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
scheduler = TranslationScheduler(db, config_manager, current_user.username)
# Check if schedule already exists
from ...models.translate import TranslationSchedule
from ....models.translate import TranslationSchedule
existing = db.query(TranslationSchedule).filter(
TranslationSchedule.job_id == job_id
).first()
@@ -129,7 +129,7 @@ async def enable_schedule(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Enable a schedule for a translation job."""
logger.info(f"[translate_routes][enable_schedule] Job: {job_id}, User: {current_user.username}")
logger.reason(f"enable_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
scheduler = TranslationScheduler(db, config_manager, current_user.username)
schedule = scheduler.set_schedule_active(job_id, is_active=True)
@@ -160,7 +160,7 @@ async def disable_schedule(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Disable a schedule for a translation job."""
logger.info(f"[translate_routes][disable_schedule] Job: {job_id}, User: {current_user.username}")
logger.reason(f"disable_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
scheduler = TranslationScheduler(db, config_manager, current_user.username)
scheduler.set_schedule_active(job_id, is_active=False)
@@ -186,7 +186,7 @@ async def delete_schedule(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Delete schedule for a translation job."""
logger.info(f"[translate_routes][delete_schedule] Job: {job_id}, User: {current_user.username}")
logger.reason(f"delete_schedule — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
scheduler = TranslationScheduler(db, config_manager, current_user.username)
scheduler.delete_schedule(job_id)
@@ -213,7 +213,7 @@ async def get_next_executions(
config_manager: ConfigManager = Depends(get_config_manager),
):
"""Preview next N executions for a job's schedule."""
logger.info(f"[translate_routes][get_next_executions] Job: {job_id}, User: {current_user.username}")
logger.reason(f"get_next_executions — Job: {job_id}, User: {current_user.username}", extra={"src": "translate_routes"})
try:
scheduler = TranslationScheduler(db, config_manager, current_user.username)
schedule = scheduler.get_schedule(job_id)

10
backend/src/app.py
View File

@@ -27,6 +27,7 @@ from .dependencies import get_task_manager, get_scheduler_service
from .core.encryption_key import ensure_encryption_key
from .core.utils.network import NetworkError
from .core.logger import logger, belief_scope
from .core.cot_logger import seed_trace_id
from .core.database import AuthSessionLocal
from .core.auth.security import get_password_hash
from .models.auth import User, Role
@@ -130,6 +131,7 @@ def ensure_initial_admin_user() -> None:
# Startup event
@app.on_event("startup")
async def startup_event():
seed_trace_id()
with belief_scope("startup_event"):
ensure_encryption_key()
ensure_initial_admin_user()
@@ -148,6 +150,7 @@ async def startup_event():
# Shutdown event
@app.on_event("shutdown")
async def shutdown_event():
seed_trace_id()
with belief_scope("shutdown_event"):
scheduler = get_scheduler_service()
scheduler.stop()
@@ -170,6 +173,7 @@ app.add_middleware(
allow_methods=["*"],
allow_headers=["*"],
)
# #endregion app_middleware
@@ -221,6 +225,12 @@ async def log_requests(request: Request, call_next):
# #endregion log_requests
# Seed trace_id on every request — MUST be the outermost middleware so trace_id
# is seeded before log_requests (or any other middleware) runs.
from .core.middleware.trace import TraceContextMiddleware
app.add_middleware(TraceContextMiddleware)
# #region API_Routes [C:3] [TYPE Block]
# @BRIEF Register all FastAPI route groups exposed by the application entrypoint.
# @RELATION DEPENDS_ON -> [FastAPI_App]

42
backend/src/core/database.py
View File

@@ -156,9 +156,9 @@ def _ensure_user_dashboard_preferences_columns(bind_engine):
for statement in alter_statements:
connection.execute(text(statement))
except Exception as migration_error:
logger.warning(
"[database][EXPLORE] Profile preference additive migration failed: %s",
migration_error,
logger.explore(
"Profile preference additive migration failed",
extra={"src": "database", "error": str(migration_error)},
)
@@ -202,9 +202,9 @@ def _ensure_user_dashboard_preferences_health_columns(bind_engine):
for statement in alter_statements:
connection.execute(text(statement))
except Exception as migration_error:
logger.warning(
"[database][EXPLORE] Profile health preference additive migration failed: %s",
migration_error,
logger.explore(
"Profile health preference additive migration failed",
extra={"src": "database", "error": str(migration_error)},
)
@@ -244,9 +244,9 @@ def _ensure_llm_validation_results_columns(bind_engine):
for statement in alter_statements:
connection.execute(text(statement))
except Exception as migration_error:
logger.warning(
"[database][EXPLORE] ValidationRecord additive migration failed: %s",
migration_error,
logger.explore(
"ValidationRecord additive migration failed",
extra={"src": "database", "error": str(migration_error)},
)
@@ -284,9 +284,9 @@ def _ensure_git_server_configs_columns(bind_engine):
for statement in alter_statements:
connection.execute(text(statement))
except Exception as migration_error:
logger.warning(
"[database][EXPLORE] GitServerConfig preference additive migration failed: %s",
migration_error,
logger.explore(
"GitServerConfig preference additive migration failed",
extra={"src": "database", "error": str(migration_error)},
)
@@ -345,9 +345,9 @@ def _ensure_auth_users_columns(bind_engine):
},
)
except Exception as migration_error:
logger.warning(
"[database][EXPLORE] Auth users additive migration failed: %s",
migration_error,
logger.explore(
"Auth users additive migration failed",
extra={"src": "database", "error": str(migration_error)},
)
raise
@@ -365,9 +365,9 @@ def ensure_connection_configs_table(bind_engine):
try:
ConnectionConfig.__table__.create(bind=bind_engine, checkfirst=True)
except Exception as migration_error:
logger.warning(
"[database][EXPLORE] ConnectionConfig table ensure failed: %s",
migration_error,
logger.explore(
"ConnectionConfig table ensure failed",
extra={"src": "database", "error": str(migration_error)},
)
raise
@@ -437,9 +437,9 @@ def _ensure_filter_source_enum_values(bind_engine):
extra={"added": missing_values},
)
except Exception as migration_error:
logger.warning(
"[database][EXPLORE] FilterSource enum additive migration failed: %s",
migration_error,
logger.explore(
"FilterSource enum additive migration failed",
extra={"src": "database", "error": str(migration_error)},
)

330
backend/src/core/logger.py
View File

@@ -1,16 +1,30 @@
# #region LoggerModule [TYPE Module] [SEMANTICS logging, websocket, streaming, handler]
# @BRIEF Configures the application's logging system, including a custom handler for buffering logs and streaming them over WebSockets.
# @LAYER: Core
# @RELATION Used by the main application and other modules to log events. The WebSocketLogHandler is used by the WebSocket endpoint in app.py.
# #region LoggerModule [C:5] [TYPE Module] [SEMANTICS logging,websocket,streaming,handler,cot,json]
# @BRIEF Application logging system with CotJsonFormatter producing molecular CoT JSON output.
# @LAYER Core
# @RELATION USED_BY -> [All application modules]
# @RELATION DEPENDS_ON -> [CotLoggerModule]
# @RELATION DEPENDS_ON -> [WebSocketLogHandler]
# @PRE Python 3.7+ with cot_logger ContextVars available.
# @POST All log output is single-line JSON with ts, level, trace_id, src, marker, intent, payload, error, span_id.
# @SIDE_EFFECT Configures global logger handlers and formatters; seeds global _enable_belief_state and _task_log_level; writes JSON to console/files.
# @RATIONALE Replaced BeliefFormatter text-based [Entry]/[Exit]/[Action] logging with CotJsonFormatter JSON output. Old format was not machine-parseable and mixed presentation with intent. CotJsonFormatter produces structured JSON consistent with cot_logger.py MarkerLogger protocol.
# @REJECTED Keeping both formatters side-by-side was rejected (two inconsistent output formats would confuse log consumers).
# @DATA_CONTRACT All log records conform to molecular CoT JSON schema: {ts, level, trace_id, src, marker, intent, span_id?, payload?, error?}
# @INVARIANT CotJsonFormatter.format() always returns valid single-line JSON string.
import json
import logging
import threading
from datetime import datetime
from typing import Dict, Any, List, Optional
from collections import deque
import types
from datetime import datetime, timezone
from contextlib import contextmanager
from logging.handlers import RotatingFileHandler
from pydantic import BaseModel, Field
from typing import Dict, Any, List, Optional
from collections import deque
from .cot_logger import _trace_id, _span_id, log as cot_log
from .ws_log_handler import WebSocketLogHandler # noqa: F401
# Thread-local storage for belief state
_belief_state = threading.local()
@@ -21,80 +35,136 @@ _enable_belief_state = True
# Global task log level filter
_task_log_level = "INFO"
# #region BeliefFormatter [TYPE Class]
# @BRIEF Custom logging formatter that adds belief state prefixes to log messages.
# #region CotJsonFormatter [C:3] [TYPE Class] [SEMANTICS logging,formatter,json,cot,protocol]
# @BRIEF JSON formatter implementing the molecular CoT logging protocol. Outputs single-line JSON with ts, level, trace_id, src, marker, intent, payload, error, span_id.
# @RELATION IMPLEMENTS -> [CotJsonFormat]
class CotJsonFormatter(logging.Formatter):
"""JSON formatter implementing the molecular CoT logging protocol.
Reads structured data from the LogRecord's extra attributes (marker, intent, payload,
error, src) set via the ``extra`` kwarg. Falls back to plain message wrapping when no
structured extra is present.
Output format (single-line JSON)::
{
"ts": "2026-05-12T10:30:00.123",
"level": "INFO",
"trace_id": "uuid-or-no-trace",
"src": "module.name",
"marker": "REASON|REFLECT|EXPLORE",
"intent": "human-readable intent",
"span_id": "optional-span",
"payload": { ... }, # optional
"error": "..." # optional
}
"""
# #region CotJsonFormatter.format [C:2] [TYPE Function] [SEMANTICS format,json,record]
# @BRIEF Format a LogRecord into a single-line CoT JSON string with molecular CoT protocol fields.
# @INVARIANT Output is always valid single-line JSON.
def format(self, record):
# Try to extract structured data from extra kwargs on the record
marker = getattr(record, 'marker', None)
intent = getattr(record, 'intent', None)
payload = getattr(record, 'payload', None)
error = getattr(record, 'error', None)
src = getattr(record, 'src', None) or record.name
# Default marker for plain messages
if not marker:
marker = "REASON"
if not intent:
intent = record.getMessage()
log_obj = {
"ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
"level": record.levelname,
"trace_id": _trace_id.get() or "no-trace",
"src": src,
"marker": marker,
"intent": intent,
}
span_id = _span_id.get()
if span_id:
log_obj["span_id"] = span_id
if payload:
log_obj["payload"] = payload
if error:
log_obj["error"] = error
return json.dumps(log_obj, ensure_ascii=False, default=str)
# #endregion CotJsonFormatter.format
# #endregion CotJsonFormatter
# #region BeliefFormatter [C:2] [TYPE Class] [SEMANTICS logging,formatter,legacy,text]
# @BRIEF Legacy formatter that adds belief state prefixes to log messages. Deprecated in favor of CotJsonFormatter.
# @RATIONALE Kept for backward compatibility; may be used by external consumers of this module.
class BeliefFormatter(logging.Formatter):
# [DEF:format:Function]
# @PURPOSE: Formats the log record, adding belief state context if available.
# @PRE: record is a logging.LogRecord.
# @POST: Returns formatted string with molecular CoT markers.
# @PARAM: record (logging.LogRecord) - The log record to format.
# @RETURN: str - The formatted log message.
# @SEMANTICS: logging, formatter, context, cot
"""Legacy formatter that adds belief state prefixes to log messages. Deprecated in favor of CotJsonFormatter."""
def format(self, record):
anchor_id = getattr(_belief_state, 'anchor_id', None)
if anchor_id:
msg = str(record.msg)
# Molecular CoT topology markers only (REASON | REFLECT | EXPLORE)
markers = ("[EXPLORE]", "[REASON]", "[REFLECT]")
# Avoid duplicating anchor or overriding explicit markers
if msg.startswith(f"[{anchor_id}]"):
pass
elif any(msg.startswith(m) for m in markers):
record.msg = f"[{anchor_id}]{msg}"
else:
# Default molecular bond
record.msg = f"[{anchor_id}][REASON] {msg}"
return super().format(record)
# [/DEF:format:Function]
# #endregion BeliefFormatter
# Re-using LogEntry from task_manager for consistency
# #region LogEntry [TYPE Class] [SEMANTICS log, entry, record, pydantic]
# @BRIEF A Pydantic model representing a single, structured log entry. This is a re-definition for consistency, as it's also defined in task_manager.py.
# #region LogEntry [C:1] [TYPE Class] [SEMANTICS log,entry,record,pydantic]
# @BRIEF A Pydantic model representing a single, structured log entry.
class LogEntry(BaseModel):
timestamp: datetime = Field(default_factory=datetime.utcnow)
level: str
message: str
context: Optional[Dict[str, Any]] = None
# #endregion LogEntry
# #region belief_scope [TYPE Function] [SEMANTICS logging, context, belief_state, cot]
# @BRIEF Context manager for Molecular CoT structured logging.
# @PRE: anchor_id must be provided.
# @POST: Thread-local belief state is updated; REASON/REFLECT/EXPLORE markers emitted.
# #region belief_scope [C:3] [TYPE Function] [SEMANTICS logging,context,belief_state,cot]
# @BRIEF Context manager for Molecular CoT structured logging. Uses cot_logger.log() for JSON output.
# @PRE anchor_id must be provided.
# @POST Thread-local belief state is updated; REASON/REFLECT/EXPLORE markers emitted via structured JSON logger.
@contextmanager
def belief_scope(anchor_id: str, message: str = ""):
# Set thread-local anchor_id FIRST so BeliefFormatter can apply prefix
old_anchor = getattr(_belief_state, 'anchor_id', None)
_belief_state.anchor_id = anchor_id
try:
# Log entry as REASON (deep-reasoning bond)
# Log entry as REASON (gated by _enable_belief_state)
if _enable_belief_state:
intent_msg = message or f"Entering {anchor_id}"
logger.reason(intent_msg)
cot_log(anchor_id, "REASON", intent_msg, level="DEBUG")
yield
# Log success as REFLECT (self-reflection bond)
logger.reflect(f"Completed successfully — {message or 'no message'}")
# Log success as REFLECT (always logged)
cot_log(anchor_id, "REFLECT", "Coherence OK", level="DEBUG")
except Exception as e:
# Log exception as EXPLORE (self-exploration bond)
logger.explore(f"Assumption violated — {str(e)}")
# Log exception as EXPLORE (always logged)
cot_log(anchor_id, "EXPLORE", f"Assumption violated", error=str(e), level="WARNING")
raise
finally:
# Restore old anchor
_belief_state.anchor_id = old_anchor
# #endregion belief_scope
# #region configure_logger [TYPE Function] [SEMANTICS logging, configuration, initialization]
# @BRIEF Configures the logger with the provided logging settings.
# @PRE: config is a valid LoggingConfig instance.
# @POST: Logger level, handlers, belief state flag, and task log level are updated.
# #region configure_logger [C:4] [TYPE Function] [SEMANTICS logging,configuration,initialization,cot]
# @BRIEF Configures the main logger and the CoT logger with CotJsonFormatter, file handlers, and global flags.
# @RELATION CALLS -> [CotJsonFormatter]
# @RELATION CALLS -> [CotLoggerModule]
# @PRE config is a valid LoggingConfig instance.
# @POST Logger levels, handlers, formatters, belief state flag, and task log level are updated.
# @SIDE_EFFECT Mutates global _enable_belief_state, _task_log_level; adds/removes handlers on both logger and cot_logger; creates file directories.
def configure_logger(config):
global _enable_belief_state, _task_log_level
_enable_belief_state = config.enable_belief_state
@@ -115,178 +185,124 @@ def configure_logger(config):
from pathlib import Path
log_file = Path(config.file_path)
log_file.parent.mkdir(parents=True, exist_ok=True)
file_handler = RotatingFileHandler(
config.file_path,
maxBytes=config.max_bytes,
backupCount=config.backup_count
)
file_handler.setFormatter(BeliefFormatter(
'[%(asctime)s][%(levelname)s][%(name)s] %(message)s'
))
file_handler.setFormatter(CotJsonFormatter())
logger.addHandler(file_handler)
# Update existing handlers' formatters to BeliefFormatter
# Update existing handlers' formatters to CotJsonFormatter
for handler in logger.handlers:
if not isinstance(handler, RotatingFileHandler):
handler.setFormatter(BeliefFormatter(
'[%(asctime)s][%(levelname)s][%(name)s] %(message)s'
))
handler.setFormatter(CotJsonFormatter())
# Configure the 'cot' logger for consistent JSON output
from .cot_logger import cot_logger
cot_logger.setLevel(level)
if not cot_logger.handlers:
cot_handler = logging.StreamHandler()
cot_handler.setFormatter(CotJsonFormatter())
cot_logger.addHandler(cot_handler)
# #endregion configure_logger
# #region get_task_log_level [TYPE Function] [SEMANTICS logging, configuration, getter]
# #region get_task_log_level [C:1] [TYPE Function] [SEMANTICS logging,configuration,getter]
# @BRIEF Returns the current task log level filter.
# @PRE: None.
# @POST: Returns the task log level string.
def get_task_log_level() -> str:
"""Returns the current task log level filter."""
return _task_log_level
# #endregion get_task_log_level
# #region should_log_task_level [TYPE Function] [SEMANTICS logging, filter, level]
# #region should_log_task_level [C:1] [TYPE Function] [SEMANTICS logging,filter,level]
# @BRIEF Checks if a log level should be recorded based on task_log_level setting.
# @PRE: level is a valid log level string.
# @POST: Returns True if level meets or exceeds task_log_level threshold.
def should_log_task_level(level: str) -> bool:
"""Checks if a log level should be recorded based on task_log_level setting."""
level_order = {"DEBUG": 0, "INFO": 1, "WARNING": 2, "ERROR": 3}
current_level = _task_log_level.upper()
check_level = level.upper()
current_order = level_order.get(current_level, 1) # Default to INFO
current_order = level_order.get(current_level, 1)
check_order = level_order.get(check_level, 1)
return check_order >= current_order
# #endregion should_log_task_level
# #region WebSocketLogHandler [TYPE Class] [SEMANTICS logging, handler, websocket, buffer]
# @BRIEF A custom logging handler that captures log records into a buffer. It is designed to be extended for real-time log streaming over WebSockets.
class WebSocketLogHandler(logging.Handler):
"""
A logging handler that stores log records and can be extended to send them
over WebSockets.
"""
# [DEF:__init__:Function]
# @PURPOSE: Initializes the handler with a fixed-capacity buffer.
# @PRE: capacity is an integer.
# @POST: Instance initialized with empty deque.
# @PARAM: capacity (int) - Maximum number of logs to keep in memory.
# @SEMANTICS: logging, initialization, buffer
def __init__(self, capacity: int = 1000):
super().__init__()
self.log_buffer: deque[LogEntry] = deque(maxlen=capacity)
# In a real implementation, you'd have a way to manage active WebSocket connections
# e.g., self.active_connections: Set[WebSocket] = set()
# [/DEF:__init__:Function]
# [DEF:emit:Function]
# @PURPOSE: Captures a log record, formats it, and stores it in the buffer.
# @PRE: record is a logging.LogRecord.
# @POST: Log is added to the log_buffer.
# @PARAM: record (logging.LogRecord) - The log record to emit.
# @SEMANTICS: logging, handler, buffer
def emit(self, record: logging.LogRecord):
try:
log_entry = LogEntry(
level=record.levelname,
message=self.format(record),
context={
"name": record.name,
"pathname": record.pathname,
"lineno": record.lineno,
"funcName": record.funcName,
"process": record.process,
"thread": record.thread,
}
)
self.log_buffer.append(log_entry)
# Here you would typically send the log_entry to all active WebSocket connections
# for real-time streaming to the frontend.
# Example: for ws in self.active_connections: await ws.send_json(log_entry.dict())
except Exception:
self.handleError(record)
# [/DEF:emit:Function]
# [DEF:get_recent_logs:Function]
# @PURPOSE: Returns a list of recent log entries from the buffer.
# @PRE: None.
# @POST: Returns list of LogEntry objects.
# @RETURN: List[LogEntry] - List of buffered log entries.
# @SEMANTICS: logging, buffer, retrieval
def get_recent_logs(self) -> List[LogEntry]:
"""
Returns a list of recent log entries from the buffer.
"""
return list(self.log_buffer)
# [/DEF:get_recent_logs:Function]
# #endregion WebSocketLogHandler
# #region Logger [TYPE Global] [SEMANTICS logger, global, instance]
# @BRIEF The global logger instance for the application, configured with both a console handler and the custom WebSocket handler.
# #region Logger [C:2] [TYPE Global] [SEMANTICS logger,global,instance]
# @BRIEF The global application logger instance configured with CotJsonFormatter, console handler, and WebSocketLogHandler.
logger = logging.getLogger("superset_tools_app")
# #region believed [TYPE Function]
# #region believed [C:2] [TYPE Function] [SEMANTICS decorator,belief,scope]
# @BRIEF A decorator that wraps a function in a belief scope.
# @PRE: anchor_id must be a string.
# @POST: Returns a decorator function.
def believed(anchor_id: str):
# [DEF:decorator:Function]
# @PURPOSE: Internal decorator for belief scope.
# @PRE: func must be a callable.
# @POST: Returns the wrapped function.
def decorator(func):
# [DEF:wrapper:Function]
# @PURPOSE: Internal wrapper that enters belief scope.
# @PRE: None.
# @POST: Executes the function within a belief scope.
def wrapper(*args, **kwargs):
with belief_scope(anchor_id):
return func(*args, **kwargs)
# [/DEF:wrapper:Function]
return wrapper
# [/DEF:decorator:Function]
return decorator
# #endregion believed
logger.setLevel(logging.INFO)
# Create a formatter
formatter = BeliefFormatter(
'[%(asctime)s][%(levelname)s][%(name)s] %(message)s'
)
# Create CotJsonFormatter
_formatter = CotJsonFormatter()
# Add console handler
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
console_handler.setFormatter(_formatter)
logger.addHandler(console_handler)
# Add WebSocket log handler
websocket_log_handler = WebSocketLogHandler()
websocket_log_handler.setFormatter(formatter)
websocket_log_handler.setFormatter(_formatter)
logger.addHandler(websocket_log_handler)
# Example usage:
# logger.info("Application started", extra={"context_key": "context_value"})
# logger.error("An error occurred", exc_info=True)
import types
# #region explore [TYPE Function] [SEMANTICS log, explore, molecule]
# @BRIEF Logs an EXPLORE message (Van der Waals force) for searching, alternatives, and hypotheses.
# #region explore [C:2] [TYPE Function] [SEMANTICS log,explore,marker,warning]
# @BRIEF Logs an EXPLORE marker (WARNING level) with structured extra data for CotJsonFormatter.
def explore(self, msg, *args, **kwargs):
self.warning(f"[EXPLORE] {msg}", *args, **kwargs)
"""Log an EXPLORE marker — searching, alternatives, violated assumptions.
Passes structured ``extra`` data so CotJsonFormatter produces proper JSON.
"""
user_extra = kwargs.pop('extra', {})
extra = {'marker': 'EXPLORE', 'intent': msg}
extra.update(user_extra)
self.warning(msg, *args, extra=extra, **kwargs)
# #endregion explore
# #region reason [TYPE Function] [SEMANTICS log, reason, molecule]
# @BRIEF Logs a REASON message (Covalent bond) for strict deduction and core logic.
# #region reason [C:2] [TYPE Function] [SEMANTICS log,reason,marker,debug]
# @BRIEF Logs a REASON marker (DEBUG level) with structured extra data for CotJsonFormatter.
def reason(self, msg, *args, **kwargs):
self.info(f"[REASON] {msg}", *args, **kwargs)
"""Log a REASON marker — strict deduction, core logic.
Passes structured ``extra`` data so CotJsonFormatter produces proper JSON.
"""
user_extra = kwargs.pop('extra', {})
extra = {'marker': 'REASON', 'intent': msg}
extra.update(user_extra)
self.debug(msg, *args, extra=extra, **kwargs)
# #endregion reason
# #region reflect [TYPE Function] [SEMANTICS log, reflect, molecule]
# @BRIEF Logs a REFLECT message (Hydrogen bond) for self-check and structural validation.
# #region reflect [C:2] [TYPE Function] [SEMANTICS log,reflect,marker,debug]
# @BRIEF Logs a REFLECT marker (DEBUG level) with structured extra data for CotJsonFormatter.
def reflect(self, msg, *args, **kwargs):
self.debug(f"[REFLECT] {msg}", *args, **kwargs)
"""Log a REFLECT marker — self-check, structural validation.
Passes structured ``extra`` data so CotJsonFormatter produces proper JSON.
"""
user_extra = kwargs.pop('extra', {})
extra = {'marker': 'REFLECT', 'intent': msg}
extra.update(user_extra)
self.debug(msg, *args, extra=extra, **kwargs)
# #endregion reflect
logger.explore = types.MethodType(explore, logger)
@@ -294,4 +310,4 @@ logger.reason = types.MethodType(reason, logger)
logger.reflect = types.MethodType(reflect, logger)
# #endregion Logger
# #endregion LoggerModule
# #endregion LoggerModule

10
backend/src/core/mapping_service.py
View File

@@ -109,8 +109,9 @@ class IdMappingService:
If incremental=True, only fetches items changed since the max last_synced_at date.
"""
with belief_scope("IdMappingService.sync_environment"):
logger.info(
f"[IdMappingService.sync_environment][REASON] Starting sync for environment {environment_id} (incremental={incremental})"
logger.reason(
f"Starting sync for environment {environment_id}",
extra={"src": "IdMappingService.sync_environment", "payload": {"environment_id": environment_id, "incremental": incremental}},
)
# Implementation Note: In a real scenario, superset_client needs to be an instance
@@ -223,8 +224,9 @@ class IdMappingService:
deleted = stale_query.delete(synchronize_session="fetch")
if deleted:
total_deleted += deleted
logger.info(
f"[IdMappingService.sync_environment][REASON] Removed {deleted} stale {endpoint} mapping(s) for {environment_id}"
logger.reason(
f"Removed {deleted} stale {endpoint} mapping(s) for {environment_id}",
extra={"src": "IdMappingService.sync_environment", "payload": {"deleted": deleted, "endpoint": endpoint, "environment_id": environment_id}},
)
except Exception as loop_e:

5
backend/src/core/migration/archive_parser.py
View File

@@ -73,9 +73,8 @@ class MigrationArchiveParser:
objects.append(normalized)
except Exception as exc:
logger.reflect(
"[MigrationArchiveParser._collect_yaml_objects][REFLECT] skip_invalid_yaml path=%s error=%s",
file_path,
exc,
f"skip_invalid_yaml path={file_path} error={exc}",
extra={"src": "MigrationArchiveParser._collect_yaml_objects"},
)
return objects

8
backend/src/core/migration/dry_run_orchestrator.py
View File

@@ -61,8 +61,9 @@ class MigrationDryRunService:
db: Session,
) -> Dict[str, Any]:
with belief_scope("MigrationDryRunService.run"):
logger.explore(
"[MigrationDryRunService.run][EXPLORE] starting dry-run pipeline"
logger.reason(
"Starting dry-run pipeline",
extra={"src": "MigrationDryRunService.run"},
)
engine = MigrationEngine()
db_mapping = (
@@ -143,7 +144,8 @@ class MigrationDryRunService:
]
logger.reason(
"[MigrationDryRunService.run][REASON] dry-run payload assembled"
"Dry-run payload assembled",
extra={"src": "MigrationDryRunService.run"},
)
return {
"generated_at": datetime.now(timezone.utc).isoformat(),

7
backend/src/core/superset_client/_base.py
View File

@@ -298,10 +298,9 @@ class SupersetClientBase:
endpoint=config["endpoint"],
pagination_options={"base_query": validated, "results_field": "result"},
)
app_logger.info(
"[get_all_resources][REFLECT] Fetched %d %s resources.",
len(data),
resource_type,
app_logger.reflect(
f"Fetched {len(data)} {resource_type} resources.",
extra={"src": "get_all_resources"},
)
return data

6
backend/src/core/superset_client/_dashboards_crud.py
View File

@@ -291,9 +291,9 @@ class SupersetDashboardsCrudMixin:
response = cast(Response, response)
self._validate_export_response(response, dashboard_id)
filename = self._resolve_export_filename(response, dashboard_id)
app_logger.info(
"[export_dashboard][REFLECT] Exported dashboard %s to %s.",
dashboard_id, filename,
app_logger.reflect(
f"Exported dashboard {dashboard_id} to {filename}.",
extra={"src": "export_dashboard"},
)
return response.content, filename

15
backend/src/core/superset_client/_dashboards_list.py
View File

@@ -39,7 +39,7 @@ class SupersetDashboardsListMixin:
},
)
total_count = len(paginated_data)
app_logger.info("[get_dashboards][REFLECT] Found %d dashboards.", total_count)
app_logger.reflect(f"Found {total_count} dashboards.", extra={"src": "get_dashboards"})
return total_count, paginated_data
# [/DEF:SupersetClientGetDashboards:Function]
@@ -125,18 +125,13 @@ class SupersetDashboardsListMixin:
if index < max_debug_samples:
app_logger.reflect(
"[REFLECT] Dashboard actor projection sample "
f"(env={getattr(self.env, 'id', None)}, dashboard_id={dash.get('id')}, "
f"raw_owners={raw_owners!r}, raw_owner_usernames={raw_owner_usernames!r}, "
f"raw_created_by={raw_created_by!r}, raw_changed_by={raw_changed_by!r}, "
f"raw_changed_by_name={raw_changed_by_name!r}, projected_owners={owners!r}, "
f"projected_created_by={projected_created_by!r}, projected_modified_by={projected_modified_by!r})"
"Dashboard actor projection sample",
extra={"src": "get_dashboards_summary", "payload": {"env": getattr(self.env, 'id', None), "dashboard_id": dash.get('id'), "raw_owners": raw_owners, "raw_owner_usernames": raw_owner_usernames, "raw_created_by": raw_created_by, "raw_changed_by": raw_changed_by, "raw_changed_by_name": raw_changed_by_name, "projected_owners": owners, "projected_created_by": projected_created_by, "projected_modified_by": projected_modified_by}},
)
app_logger.reflect(
"[REFLECT] Dashboard actor projection summary "
f"(env={getattr(self.env, 'id', None)}, dashboards={len(result)}, "
f"sampled={min(len(result), max_debug_samples)})"
"Dashboard actor projection summary",
extra={"src": "get_dashboards_summary", "payload": {"env": getattr(self.env, 'id', None), "dashboards": len(result), "sampled": min(len(result), max_debug_samples)}},
)
return result

4
backend/src/core/superset_client/_databases.py
View File

@@ -33,7 +33,7 @@ class SupersetDatabasesMixin:
},
)
total_count = len(paginated_data)
app_logger.info("[get_databases][REFLECT] Found %d databases.", total_count)
app_logger.reflect(f"Found {total_count} databases.", extra={"src": "get_databases"})
return total_count, paginated_data
# [/DEF:SupersetClientGetDatabases:Function]
@@ -49,7 +49,7 @@ class SupersetDatabasesMixin:
method="GET", endpoint=f"/database/{database_id}"
)
response = cast(Dict, response)
app_logger.info("[get_database][REFLECT] Got database %s.", database_id)
app_logger.reflect(f"Got database {database_id}.", extra={"src": "get_database"})
return response
# [/DEF:SupersetClientGetDatabase:Function]

11
backend/src/core/superset_client/_datasets.py
View File

@@ -32,7 +32,7 @@ class SupersetDatasetsMixin:
},
)
total_count = len(paginated_data)
app_logger.info("[get_datasets][REFLECT] Found %d datasets.", total_count)
app_logger.reflect(f"Found {total_count} datasets.", extra={"src": "get_datasets"})
return total_count, paginated_data
# [/DEF:SupersetClientGetDatasets:Function]
@@ -168,8 +168,9 @@ class SupersetDatasetsMixin:
"changed_on": dataset.get("changed_on"),
}
app_logger.info(
f"[get_dataset_detail][REFLECT] Got dataset {dataset_id} with {len(column_info)} columns and {len(linked_dashboards)} linked dashboards"
app_logger.reflect(
f"Got dataset {dataset_id} with {len(column_info)} columns and {len(linked_dashboards)} linked dashboards",
extra={"src": "get_dataset_detail"},
)
return result
@@ -186,7 +187,7 @@ class SupersetDatasetsMixin:
method="GET", endpoint=f"/dataset/{dataset_id}"
)
response = cast(Dict, response)
app_logger.info("[get_dataset][REFLECT] Got dataset %s.", dataset_id)
app_logger.reflect(f"Got dataset {dataset_id}.", extra={"src": "get_dataset"})
return response
# [/DEF:SupersetClientGetDataset:Function]
@@ -206,7 +207,7 @@ class SupersetDatasetsMixin:
headers={"Content-Type": "application/json"},
)
response = cast(Dict, response)
app_logger.info("[update_dataset][REFLECT] Updated dataset %s.", dataset_id)
app_logger.reflect(f"Updated dataset {dataset_id}.", extra={"src": "update_dataset"})
return response
# [/DEF:SupersetClientUpdateDataset:Function]

38
backend/src/core/superset_profile_lookup.py
View File

@@ -63,14 +63,12 @@ class SupersetAccountLookupAdapter:
query["filters"] = [{"col": "username", "opr": "ct", "value": normalized_search}]
logger.reason(
"[REASON] Lookup Superset users "
f"(env={self.environment_id}, page={normalized_page_index}, page_size={normalized_page_size})"
"Lookup Superset users",
extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "page": normalized_page_index, "page_size": normalized_page_size}},
)
logger.reflect(
"[REFLECT] Prepared Superset users lookup query "
f"(env={self.environment_id}, order_column={normalized_sort_column}, "
f"normalized_sort_order={normalized_sort_order}, "
f"payload_order_direction={query.get('order_direction')})"
"Prepared Superset users lookup query",
extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "order_column": normalized_sort_column, "sort_order": normalized_sort_order, "direction": query.get("order_direction")}},
)
primary_error: Optional[Exception] = None
@@ -78,8 +76,8 @@ class SupersetAccountLookupAdapter:
for attempt_index, endpoint in enumerate(("/security/users/", "/security/users"), start=1):
try:
logger.reason(
"[REASON] Users lookup request attempt "
f"(env={self.environment_id}, attempt={attempt_index}, endpoint={endpoint})"
"Users lookup request attempt",
extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "attempt": attempt_index, "endpoint": endpoint}},
)
response = self.network_client.request(
method="GET",
@@ -87,8 +85,8 @@ class SupersetAccountLookupAdapter:
params={"q": json.dumps(query)},
)
logger.reflect(
"[REFLECT] Users lookup endpoint succeeded "
f"(env={self.environment_id}, attempt={attempt_index}, endpoint={endpoint})"
"Users lookup endpoint succeeded",
extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "attempt": attempt_index, "endpoint": endpoint}},
)
return self._normalize_lookup_payload(
response=response,
@@ -103,10 +101,8 @@ class SupersetAccountLookupAdapter:
cause_response = getattr(cause, "response", None)
status_code = getattr(cause_response, "status_code", None)
logger.explore(
"[EXPLORE] Users lookup endpoint failed "
f"(env={self.environment_id}, attempt={attempt_index}, endpoint={endpoint}, "
f"error_type={type(exc).__name__}, status_code={status_code}, "
f"payload_order_direction={query.get('order_direction')}): {exc}"
"Users lookup endpoint failed",
extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "attempt": attempt_index, "endpoint": endpoint, "error_type": type(exc).__name__, "status_code": status_code, "direction": query.get("order_direction")}, "error": str(exc)},
)
if last_error is not None:
@@ -119,15 +115,13 @@ class SupersetAccountLookupAdapter:
):
selected_error = primary_error
logger.reflect(
"[REFLECT] Preserving primary lookup failure over fallback auth error "
f"(env={self.environment_id}, primary_error_type={type(primary_error).__name__}, "
f"fallback_error_type={type(last_error).__name__})"
"Preserving primary lookup failure over fallback auth error",
extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "primary_error_type": type(primary_error).__name__, "fallback_error_type": type(last_error).__name__}},
)
logger.explore(
"[EXPLORE] All Superset users lookup endpoints failed "
f"(env={self.environment_id}, payload_order_direction={query.get('order_direction')}, "
f"selected_error_type={type(selected_error).__name__})"
"All Superset users lookup endpoints failed",
extra={"src": "SupersetAccountLookupAdapter.get_users_page", "payload": {"env": self.environment_id, "direction": query.get("order_direction"), "selected_error_type": type(selected_error).__name__}},
)
raise selected_error
raise SupersetAPIError("Superset users lookup failed without explicit error")
@@ -180,8 +174,8 @@ class SupersetAccountLookupAdapter:
normalized_items.append(candidate)
logger.reflect(
"[REFLECT] Normalized lookup payload "
f"(env={self.environment_id}, items={len(normalized_items)}, total={max(total, len(normalized_items))})"
"Normalized lookup payload",
extra={"src": "SupersetAccountLookupAdapter._normalize_lookup_payload", "payload": {"env": self.environment_id, "items": len(normalized_items), "total": max(total, len(normalized_items))}},
)
return {

60
backend/src/core/task_manager/persistence.py
View File

@@ -57,20 +57,20 @@ class TaskPersistenceService:
# @POST: Returns parsed JSON object, list, string, or primitive
@staticmethod
def _json_load_if_needed(value):
with belief_scope("TaskPersistenceService._json_load_if_needed"):
if value is None:
return None
if isinstance(value, (dict, list)):
return value
if isinstance(value, str):
stripped = value.strip()
if stripped == "" or stripped.lower() == "null":
return None
try:
return json.loads(stripped)
except json.JSONDecodeError:
return value
# Hot-path utility — no entry/exit logging to reduce noise
if value is None:
return None
if isinstance(value, (dict, list)):
return value
if isinstance(value, str):
stripped = value.strip()
if stripped == "" or stripped.lower() == "null":
return None
try:
return json.loads(stripped)
except json.JSONDecodeError:
return value
return value
# [/DEF:_json_load_if_needed:Function]
@@ -81,15 +81,15 @@ class TaskPersistenceService:
# @POST: Returns datetime object or None
@staticmethod
def _parse_datetime(value):
with belief_scope("TaskPersistenceService._parse_datetime"):
if value is None or isinstance(value, datetime):
return value
if isinstance(value, str):
try:
return datetime.fromisoformat(value)
except ValueError:
return None
return None
# Hot-path utility — no entry/exit logging to reduce startup noise
if value is None or isinstance(value, datetime):
return value
if isinstance(value, str):
try:
return datetime.fromisoformat(value)
except ValueError:
return None
return None
# [/DEF:_parse_datetime:Function]
@@ -188,14 +188,14 @@ class TaskPersistenceService:
# Ensure params and result are JSON serializable
def json_serializable(obj):
with belief_scope("TaskPersistenceService.json_serializable"):
if isinstance(obj, dict):
return {k: json_serializable(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [json_serializable(v) for v in obj]
elif isinstance(obj, datetime):
return obj.isoformat()
return obj
# Hot-path utility — no entry/exit logging
if isinstance(obj, dict):
return {k: json_serializable(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [json_serializable(v) for v in obj]
elif isinstance(obj, datetime):
return obj.isoformat()
return obj
record.params = json_serializable(task.params)
record.result = json_serializable(task.result)

2
backend/src/core/utils/async_network.py
View File

@@ -149,7 +149,7 @@ class AsyncAPIClient:
}
self._authenticated = True
SupersetAuthCache.set(self._auth_cache_key, self._tokens)
app_logger.info("[async_authenticate][REFLECT] Authenticated successfully.")
app_logger.reflect("Authenticated successfully.", extra={"src": "async_authenticate"})
return self._tokens
except httpx.HTTPStatusError as exc:
SupersetAuthCache.invalidate(self._auth_cache_key)

4
backend/src/core/utils/fileio.py
View File

@@ -81,7 +81,7 @@ def remove_empty_directories(root_dir: str) -> int:
app_logger.info("[remove_empty_directories][State] Removed empty directory: %s", current_dir)
except OSError as e:
app_logger.error("[remove_empty_directories][Failure] Failed to remove %s: %s", current_dir, e)
app_logger.info("[remove_empty_directories][REFLECT] Removed %d empty directories.", removed_count)
app_logger.reflect(f"Removed {removed_count} empty directories.", extra={"src": "remove_empty_directories"})
return removed_count
# #endregion remove_empty_directories
@@ -354,7 +354,7 @@ def create_dashboard_export(zip_path: Union[str, Path], source_paths: List[Union
if item.is_file() and item.suffix.lower() not in exclude_ext:
arcname = item.relative_to(src_path.parent)
zipf.write(item, arcname)
app_logger.info("[create_dashboard_export][REFLECT] Archive created: %s", zip_path)
app_logger.reflect(f"Archive created: {zip_path}", extra={"src": "create_dashboard_export"})
return True
except (IOError, zipfile.BadZipFile, AssertionError) as e:
app_logger.error("[create_dashboard_export][Failure] Error: %s", e, exc_info=True)

6
backend/src/core/utils/network.py
View File

@@ -162,7 +162,7 @@ class APIClient:
# @POST: APIClient instance is initialized with a session.
def __init__(self, config: Dict[str, Any], verify_ssl: bool = True, timeout: int = DEFAULT_TIMEOUT):
with belief_scope("__init__"):
app_logger.info("[APIClient.__init__][REASON] Initializing APIClient.")
app_logger.reason("Initializing APIClient.", extra={"src": "APIClient.__init__"})
self.base_url: str = self._normalize_base_url(config.get("base_url", ""))
self.api_base_url: str = f"{self.base_url}/api/v1"
self.auth = config.get("auth")
@@ -175,7 +175,7 @@ class APIClient:
verify_ssl,
)
self._authenticated = False
app_logger.info("[APIClient.__init__][REFLECT] APIClient initialized.")
app_logger.reflect("APIClient initialized.", extra={"src": "APIClient.__init__"})
# [/DEF:APIClient.__init__:Function]
# [DEF:_init_session:Function]
@@ -291,7 +291,7 @@ class APIClient:
self._tokens = {"access_token": access_token, "csrf_token": csrf_response.json()["result"]}
self._authenticated = True
SupersetAuthCache.set(self._auth_cache_key, self._tokens)
app_logger.info("[authenticate][REFLECT] Authenticated successfully.")
app_logger.reflect("Authenticated successfully.", extra={"src": "authenticate"})
return self._tokens
except requests.exceptions.HTTPError as e:
SupersetAuthCache.invalidate(self._auth_cache_key)

18
backend/src/plugins/git_plugin.py
View File

@@ -134,7 +134,7 @@ class GitPlugin(PluginBase):
# @POST: Плагин готов к выполнению задач.
async def initialize(self):
with belief_scope("GitPlugin.initialize"):
app_logger.info("[GitPlugin.initialize][REASON] Initializing Git Integration Plugin logic.")
app_logger.reason("Initializing Git Integration Plugin logic.", extra={"src": "GitPlugin.initialize"})
# [DEF:execute:Function]
# @PURPOSE: Основной метод выполнения задач плагина с поддержкой TaskContext.
@@ -245,9 +245,9 @@ class GitPlugin(PluginBase):
# 5. Автоматический staging изменений (не коммит, чтобы юзер мог проверить diff)
try:
repo.git.add(A=True)
app_logger.info("[_handle_sync][REASON] Changes staged in git")
app_logger.reason("Changes staged in git", extra={"src": "_handle_sync"})
except Exception as ge:
app_logger.warning(f"[_handle_sync][REASON] Failed to stage changes: {ge}")
app_logger.explore(f"Failed to stage changes: {ge}", extra={"src": "_handle_sync"})
app_logger.info(f"[_handle_sync][Coherence:OK] Dashboard {dashboard_id} synced successfully.")
return {"status": "success", "message": "Dashboard synced and flattened in local repository"}
@@ -315,7 +315,7 @@ class GitPlugin(PluginBase):
f.write(zip_buffer.getvalue())
try:
app_logger.info(f"[_handle_deploy][REASON] Importing dashboard to {env.name}")
app_logger.reason(f"Importing dashboard to {env.name}", extra={"src": "_handle_deploy"})
result = client.import_dashboard(temp_zip_path)
app_logger.info(f"[_handle_deploy][Coherence:OK] Deployment successful for dashboard {dashboard_id}.")
return {"status": "success", "message": f"Dashboard deployed to {env.name}", "details": result}
@@ -336,13 +336,13 @@ class GitPlugin(PluginBase):
# @RETURN: Environment - Объект конфигурации окружения.
def _get_env(self, env_id: Optional[str] = None):
with belief_scope("GitPlugin._get_env"):
app_logger.info(f"[_get_env][REASON] Fetching environment for ID: {env_id}")
app_logger.reason(f"Fetching environment for ID: {env_id}", extra={"src": "_get_env"})
# Priority 1: ConfigManager (config.json)
if env_id:
env = self.config_manager.get_environment(env_id)
if env:
app_logger.info(f"[_get_env][REFLECT] Found environment by ID in ConfigManager: {env.name}")
app_logger.reflect(f"Found environment by ID in ConfigManager: {env.name}", extra={"src": "_get_env"})
return env
# Priority 2: Database (DeploymentEnvironment)
@@ -360,7 +360,7 @@ class GitPlugin(PluginBase):
db_env = db.query(DeploymentEnvironment).first()
if db_env:
app_logger.info(f"[_get_env][REFLECT] Found environment in DB: {db_env.name}")
app_logger.reflect(f"Found environment in DB: {db_env.name}", extra={"src": "_get_env"})
from src.core.config_models import Environment
# Use token as password for SupersetClient
return Environment(
@@ -382,11 +382,11 @@ class GitPlugin(PluginBase):
# but we have other envs, maybe it's one of them?
env = next((e for e in envs if e.id == env_id), None)
if env:
app_logger.info(f"[_get_env][REFLECT] Found environment {env_id} in ConfigManager list")
app_logger.reflect(f"Found environment {env_id} in ConfigManager list", extra={"src": "_get_env"})
return env
if not env_id:
app_logger.info(f"[_get_env][REFLECT] Using first environment from ConfigManager: {envs[0].name}")
app_logger.reflect(f"Using first environment from ConfigManager: {envs[0].name}", extra={"src": "_get_env"})
return envs[0]
app_logger.error(f"[_get_env][Coherence:Failed] No environments configured (searched config.json and DB). env_id={env_id}")

11
backend/src/plugins/storage/plugin.py
View File

@@ -186,7 +186,7 @@ class StoragePlugin(PluginBase):
# Use singular name for consistency with BackupPlugin and GitService
path = root / category.value
path.mkdir(parents=True, exist_ok=True)
logger.debug(f"[StoragePlugin][REASON] Ensured directory: {path}")
logger.reason(f"Ensured directory: {path}", extra={"src": "StoragePlugin.ensure_directories"})
# [/DEF:ensure_directories:Function]
# [DEF:validate_path:Function]
@@ -221,8 +221,9 @@ class StoragePlugin(PluginBase):
) -> List[StoredFile]:
with belief_scope("StoragePlugin:list_files"):
root = self.get_storage_root()
logger.info(
f"[StoragePlugin][REASON] Listing files in root: {root}, category: {category}, subpath: {subpath}, recursive: {recursive}"
logger.reason(
f"Listing files in root: {root}",
extra={"src": "StoragePlugin.list_files", "payload": {"root": str(root), "category": category.value if category else None, "subpath": subpath, "recursive": recursive}},
)
files = []
@@ -258,7 +259,7 @@ class StoragePlugin(PluginBase):
if not target_dir.exists():
continue
logger.debug(f"[StoragePlugin][REASON] Scanning directory: {target_dir}")
logger.reason(f"Scanning directory: {target_dir}", extra={"src": "StoragePlugin.list_files"})
if recursive:
for current_root, dirs, filenames in os.walk(target_dir):
@@ -358,7 +359,7 @@ class StoragePlugin(PluginBase):
shutil.rmtree(full_path)
else:
full_path.unlink()
logger.info(f"[StoragePlugin][REASON] Deleted: {full_path}")
logger.reason(f"Deleted: {full_path}", extra={"src": "StoragePlugin.delete_file"})
else:
raise FileNotFoundError(f"Item {path} not found")
# [/DEF:delete_file:Function]

10
backend/src/scripts/migrate_sqlite_to_postgres.py
View File

@@ -164,8 +164,9 @@ def _ensure_target_schema(engine) -> None:
def _migrate_config(engine, legacy_config_path: Optional[Path]) -> int:
with belief_scope("_migrate_config"):
if legacy_config_path is None:
logger.info(
"[_migrate_config][REASON] No legacy config.json found, skipping"
logger.reason(
"No legacy config.json found, skipping",
extra={"src": "_migrate_config"},
)
return 0
@@ -327,8 +328,9 @@ def run_migration(
sqlite_path: Path, target_url: str, legacy_config_path: Optional[Path]
) -> Dict[str, int]:
with belief_scope("run_migration"):
logger.info(
"[run_migration][REASON] sqlite=%s target=%s", sqlite_path, target_url
logger.reason(
f"sqlite={sqlite_path} target={target_url}",
extra={"src": "run_migration"},
)
if not sqlite_path.exists():
raise FileNotFoundError(f"SQLite source not found: {sqlite_path}")

26
backend/src/scripts/seed_superset_load_test.py
View File

@@ -157,8 +157,9 @@ def _resolve_target_envs(env_ids: List[str]) -> Dict[str, Environment]:
env = Environment(**row)
configured[env.id] = env
except Exception as exc:
logger.warning(
f"[REFLECT] Failed loading environments from {config_path}: {exc}"
logger.reflect(
f"Failed loading environments from {config_path}: {exc}",
extra={"src": "_resolve_target_envs"},
)
for env_id in env_ids:
@@ -271,8 +272,9 @@ def seed_superset_load_data(args: argparse.Namespace) -> Dict:
templates_by_env[env_id] = _build_chart_template_pool(
client, args.template_pool_size, rng
)
logger.info(
f"[REASON] Environment {env_id}: loaded {len(templates_by_env[env_id])} chart templates"
logger.reason(
f"Environment {env_id}: loaded {len(templates_by_env[env_id])} chart templates",
extra={"src": "seed_superset_load_data"},
)
errors = 0
@@ -285,8 +287,9 @@ def seed_superset_load_data(args: argparse.Namespace) -> Dict:
dashboard_title = _generate_unique_name("lt_dash", used_dashboard_names, rng)
if args.dry_run:
logger.info(
f"[REFLECT] Dry-run dashboard create: env={env_id}, title={dashboard_title}"
logger.reflect(
f"Dry-run dashboard create: env={env_id}, title={dashboard_title}",
extra={"src": "seed_superset_load_data"},
)
continue
@@ -301,7 +304,7 @@ def seed_superset_load_data(args: argparse.Namespace) -> Dict:
created_dashboards[env_id].append(dashboard_id)
except Exception as exc:
errors += 1
logger.error(f"[EXPLORE] Failed creating dashboard in {env_id}: {exc}")
logger.explore(f"Failed creating dashboard in {env_id}: {exc}", extra={"src": "seed_superset_load_data"})
if errors >= args.max_errors:
raise RuntimeError(
f"Stopping due to max errors reached ({errors})"
@@ -351,10 +354,10 @@ def seed_superset_load_data(args: argparse.Namespace) -> Dict:
created_charts[env_id].append(chart_id)
if (index + 1) % 500 == 0:
logger.info(f"[REASON] Created {index + 1}/{args.charts} charts")
logger.reason(f"Created {index + 1}/{args.charts} charts", extra={"src": "seed_superset_load_data"})
except Exception as exc:
errors += 1
logger.error(f"[EXPLORE] Failed creating chart in {env_id}: {exc}")
logger.explore(f"Failed creating chart in {env_id}: {exc}", extra={"src": "seed_superset_load_data"})
if errors >= args.max_errors:
raise RuntimeError(
f"Stopping due to max errors reached ({errors})"
@@ -381,8 +384,9 @@ def main() -> None:
with belief_scope("seed_superset_load_test.main"):
args = _parse_args()
result = seed_superset_load_data(args)
logger.info(
f"[REFLECT] Result summary: {json.dumps(result, ensure_ascii=True)}"
logger.reflect(
f"Result summary: {json.dumps(result, ensure_ascii=True)}",
extra={"src": "main"},
)

10
backend/src/services/clean_release/approval_service.py
View File

@@ -99,7 +99,7 @@ def approve_candidate(
) -> ApprovalDecision:
with belief_scope("approval_service.approve_candidate"):
logger.reason(
f"[REASON] Evaluating approve gate candidate_id={candidate_id} report_id={report_id}"
f"Evaluating approve gate candidate_id={candidate_id} report_id={report_id}"
)
if not decided_by or not decided_by.strip():
@@ -135,7 +135,7 @@ def approve_candidate(
raise
except Exception as exc: # noqa: BLE001
logger.explore(
f"[EXPLORE] Candidate transition to APPROVED failed candidate_id={candidate_id}: {exc}"
f"Candidate transition to APPROVED failed candidate_id={candidate_id}: {exc}"
)
raise ApprovalGateError(str(exc)) from exc
@@ -153,7 +153,7 @@ def approve_candidate(
candidate_id, "APPROVED", repository=repository, actor=decided_by
)
logger.reflect(
f"[REFLECT] Approval persisted candidate_id={candidate_id} decision_id={decision.id}"
f"Approval persisted candidate_id={candidate_id} decision_id={decision.id}"
)
return decision
@@ -175,7 +175,7 @@ def reject_candidate(
) -> ApprovalDecision:
with belief_scope("approval_service.reject_candidate"):
logger.reason(
f"[REASON] Evaluating reject decision candidate_id={candidate_id} report_id={report_id}"
f"Evaluating reject decision candidate_id={candidate_id} report_id={report_id}"
)
if not decided_by or not decided_by.strip():
@@ -201,7 +201,7 @@ def reject_candidate(
candidate_id, "REJECTED", repository=repository, actor=decided_by
)
logger.reflect(
f"[REFLECT] Rejection persisted candidate_id={candidate_id} decision_id={decision.id}"
f"Rejection persisted candidate_id={candidate_id} decision_id={decision.id}"
)
return decision

16
backend/src/services/clean_release/audit_service.py
View File

@@ -21,8 +21,8 @@ def _append_event(repository, payload: Dict[str, Any]) -> None:
def audit_preparation(
candidate_id: str, status: str, repository=None, actor: str = "system"
) -> None:
logger.info(
f"[REASON] clean-release preparation candidate={candidate_id} status={status}"
logger.reason(
f"clean-release preparation candidate={candidate_id} status={status}"
)
_append_event(
repository,
@@ -45,8 +45,8 @@ def audit_check_run(
candidate_id: Optional[str] = None,
actor: str = "system",
) -> None:
logger.info(
f"[REFLECT] clean-release check_run={check_run_id} final_status={final_status}"
logger.reflect(
f"clean-release check_run={check_run_id} final_status={final_status}"
)
_append_event(
repository,
@@ -71,8 +71,8 @@ def audit_violation(
candidate_id: Optional[str] = None,
actor: str = "system",
) -> None:
logger.info(
f"[EXPLORE] clean-release violation run_id={run_id} stage={stage_name} code={code}"
logger.explore(
f"clean-release violation run_id={run_id} stage={stage_name} code={code}"
)
_append_event(
repository,
@@ -97,8 +97,8 @@ def audit_report(
run_id: Optional[str] = None,
actor: str = "system",
) -> None:
logger.info(
f"[EXPLORE] clean-release report_id={report_id} candidate={candidate_id}"
logger.explore(
f"clean-release report_id={report_id} candidate={candidate_id}"
)
_append_event(
repository,

4
backend/src/services/clean_release/compliance_execution_service.py
View File

@@ -204,7 +204,7 @@ class ComplianceExecutionService:
setattr(run, "report_id", str(getattr(report, "id")))
self.repository.save_check_run(run)
belief_logger.reflect(
f"[REFLECT] Compliance run completed run_id={getattr(run, 'id')} final_status={getattr(run, 'final_status', None)}"
f"Compliance run completed run_id={getattr(run, 'id')} final_status={getattr(run, 'final_status', None)}"
)
except Exception as exc: # noqa: BLE001
setattr(run, "status", RunStatus.FAILED.value)
@@ -213,7 +213,7 @@ class ComplianceExecutionService:
setattr(run, "finished_at", datetime.now(timezone.utc))
self.repository.save_check_run(run)
belief_logger.explore(
f"[EXPLORE] Compliance run failed run_id={getattr(run, 'id')}: {exc}"
f"Compliance run failed run_id={getattr(run, 'id')}: {exc}"
)
return ComplianceExecutionResult(

10
backend/src/services/clean_release/publication_service.py
View File

@@ -99,7 +99,7 @@ def publish_candidate(
) -> PublicationRecord:
with belief_scope("publication_service.publish_candidate"):
logger.reason(
f"[REASON] Evaluating publish gate candidate_id={candidate_id} report_id={report_id}"
f"Evaluating publish gate candidate_id={candidate_id} report_id={report_id}"
)
if not published_by or not published_by.strip():
@@ -137,7 +137,7 @@ def publish_candidate(
repository.save_candidate(candidate)
except Exception as exc: # noqa: BLE001
logger.explore(
f"[EXPLORE] Candidate transition to PUBLISHED failed candidate_id={candidate_id}: {exc}"
f"Candidate transition to PUBLISHED failed candidate_id={candidate_id}: {exc}"
)
raise PublicationGateError(str(exc)) from exc
@@ -156,7 +156,7 @@ def publish_candidate(
candidate_id, "PUBLISHED", repository=repository, actor=published_by
)
logger.reflect(
f"[REFLECT] Publication persisted candidate_id={candidate_id} publication_id={record.id}"
f"Publication persisted candidate_id={candidate_id} publication_id={record.id}"
)
return record
@@ -177,7 +177,7 @@ def revoke_publication(
) -> PublicationRecord:
with belief_scope("publication_service.revoke_publication"):
logger.reason(
f"[REASON] Evaluating revoke gate publication_id={publication_id}"
f"Evaluating revoke gate publication_id={publication_id}"
)
if not revoked_by or not revoked_by.strip():
@@ -204,7 +204,7 @@ def revoke_publication(
repository=repository,
actor=revoked_by,
)
logger.reflect(f"[REFLECT] Publication revoked publication_id={publication_id}")
logger.reflect(f"Publication revoked publication_id={publication_id}")
return record

10
backend/src/services/git/_base.py
View File

@@ -133,7 +133,7 @@ class GitServiceBase:
if source_abs == target_abs:
return source_abs
if os.path.exists(target_abs):
logger.warning(f"[_migrate_repo_directory][REASON] Target already exists, keeping source path: {target_abs}")
logger.reason(f"Target already exists, keeping source path: {target_abs}", extra={"src": "_migrate_repo_directory"})
return source_abs
Path(target_abs).parent.mkdir(parents=True, exist_ok=True)
try:
@@ -210,11 +210,11 @@ class GitServiceBase:
else:
auth_url = remote_url
if os.path.exists(repo_path):
logger.info(f"[init_repo][REASON] Opening existing repo at {repo_path}")
logger.reason(f"Opening existing repo at {repo_path}", extra={"src": "init_repo"})
try:
repo = Repo(repo_path)
except (InvalidGitRepositoryError, NoSuchPathError):
logger.warning(f"[init_repo][REASON] Existing path is not a Git repository, recreating: {repo_path}")
logger.reason(f"Existing path is not a Git repository, recreating: {repo_path}", extra={"src": "init_repo"})
stale_path = Path(repo_path)
if stale_path.exists():
shutil.rmtree(stale_path, ignore_errors=True)
@@ -226,7 +226,7 @@ class GitServiceBase:
repo = Repo.clone_from(auth_url, repo_path)
self._ensure_gitflow_branches(repo, dashboard_id)
return repo
logger.info(f"[init_repo][REASON] Cloning {remote_url} to {repo_path}")
logger.reason(f"Cloning {remote_url} to {repo_path}", extra={"src": "init_repo"})
repo = Repo.clone_from(auth_url, repo_path)
self._ensure_gitflow_branches(repo, dashboard_id)
return repo
@@ -307,7 +307,7 @@ class GitServiceBase:
with repo.config_writer(config_level="repository") as config_writer:
config_writer.set_value("user", "name", normalized_username)
config_writer.set_value("user", "email", normalized_email)
logger.info("[configure_identity][REASON] Applied repository-local git identity for dashboard %s", dashboard_id)
logger.reason(f"Applied repository-local git identity for dashboard {dashboard_id}", extra={"src": "configure_identity"})
except Exception as e:
logger.error(f"[configure_identity][Coherence:Failed] Failed to configure git identity: {e}")
raise HTTPException(status_code=500, detail=f"Failed to configure git identity: {str(e)}")

48
backend/src/services/git/_branch.py
View File

@@ -31,25 +31,28 @@ class GitServiceBranchMixin:
if "main" in local_heads:
base_commit = local_heads["main"].commit
if base_commit is None:
logger.warning(
f"[_ensure_gitflow_branches][REASON] Skipping branch bootstrap for dashboard {dashboard_id}: repository has no commits"
logger.reason(
f"Skipping branch bootstrap for dashboard {dashboard_id}: repository has no commits",
extra={"src": "_ensure_gitflow_branches"},
)
return
if "main" not in local_heads:
local_heads["main"] = repo.create_head("main", base_commit)
logger.info(f"[_ensure_gitflow_branches][REASON] Created local branch main for dashboard {dashboard_id}")
logger.reason(f"Created local branch main for dashboard {dashboard_id}", extra={"src": "_ensure_gitflow_branches"})
for branch_name in ("dev", "preprod"):
if branch_name in local_heads:
continue
local_heads[branch_name] = repo.create_head(branch_name, local_heads["main"].commit)
logger.info(
f"[_ensure_gitflow_branches][REASON] Created local branch {branch_name} for dashboard {dashboard_id}"
logger.reason(
f"Created local branch {branch_name} for dashboard {dashboard_id}",
extra={"src": "_ensure_gitflow_branches"},
)
try:
origin = repo.remote(name="origin")
except ValueError:
logger.info(
f"[_ensure_gitflow_branches][REASON] Remote origin is not configured for dashboard {dashboard_id}; skipping remote branch creation"
logger.reason(
f"Remote origin is not configured for dashboard {dashboard_id}; skipping remote branch creation",
extra={"src": "_ensure_gitflow_branches"},
)
return
remote_branch_names = set()
@@ -60,14 +63,15 @@ class GitServiceBranchMixin:
if remote_head:
remote_branch_names.add(str(remote_head))
except Exception as e:
logger.warning(f"[_ensure_gitflow_branches][REASON] Failed to fetch origin refs: {e}")
logger.reason(f"Failed to fetch origin refs: {e}", extra={"src": "_ensure_gitflow_branches"})
for branch_name in required_branches:
if branch_name in remote_branch_names:
continue
try:
origin.push(refspec=f"{branch_name}:{branch_name}")
logger.info(
f"[_ensure_gitflow_branches][REASON] Pushed branch {branch_name} to origin for dashboard {dashboard_id}"
logger.reason(
f"Pushed branch {branch_name} to origin for dashboard {dashboard_id}",
extra={"src": "_ensure_gitflow_branches"},
)
except Exception as e:
logger.error(f"[_ensure_gitflow_branches][Coherence:Failed] Failed to push branch {branch_name} to origin: {e}")
@@ -77,9 +81,9 @@ class GitServiceBranchMixin:
)
try:
repo.git.checkout("dev")
logger.info(f"[_ensure_gitflow_branches][REASON] Checked out default branch dev for dashboard {dashboard_id}")
logger.reason(f"Checked out default branch dev for dashboard {dashboard_id}", extra={"src": "_ensure_gitflow_branches"})
except Exception as e:
logger.warning(f"[_ensure_gitflow_branches][REASON] Could not checkout dev branch for dashboard {dashboard_id}: {e}")
logger.reason(f"Could not checkout dev branch for dashboard {dashboard_id}: {e}", extra={"src": "_ensure_gitflow_branches"})
# [/DEF:_ensure_gitflow_branches:Function]
# [DEF:list_branches:Function]
@@ -90,7 +94,7 @@ class GitServiceBranchMixin:
def list_branches(self, dashboard_id: int) -> List[dict]:
with belief_scope("GitService.list_branches"):
repo = self.get_repo(dashboard_id)
logger.info(f"[list_branches][REASON] Listing branches for {dashboard_id}. Refs: {repo.refs}")
logger.reason(f"Listing branches for {dashboard_id}. Refs: {repo.refs}", extra={"src": "list_branches"})
branches = []
for ref in repo.refs:
try:
@@ -104,7 +108,7 @@ class GitServiceBranchMixin:
"last_updated": datetime.fromtimestamp(ref.commit.committed_date) if hasattr(ref, 'commit') else datetime.utcnow()
})
except Exception as e:
logger.warning(f"[list_branches][REASON] Skipping ref {ref}: {e}")
logger.reason(f"Skipping ref {ref}: {e}", extra={"src": "list_branches"})
try:
active_name = repo.active_branch.name
if not any(b['name'] == active_name for b in branches):
@@ -113,7 +117,7 @@ class GitServiceBranchMixin:
"is_remote": False, "last_updated": datetime.utcnow()
})
except Exception as e:
logger.warning(f"[list_branches][REASON] Could not determine active branch: {e}")
logger.reason(f"Could not determine active branch: {e}", extra={"src": "list_branches"})
if not branches:
branches.append({
"name": "dev", "commit_hash": "0000000",
@@ -131,9 +135,9 @@ class GitServiceBranchMixin:
def create_branch(self, dashboard_id: int, name: str, from_branch: str = "main"):
with belief_scope("GitService.create_branch"):
repo = self.get_repo(dashboard_id)
logger.info(f"[create_branch][REASON] Creating branch {name} from {from_branch}")
logger.reason(f"Creating branch {name} from {from_branch}", extra={"src": "create_branch"})
if not repo.heads and not repo.remotes:
logger.warning("[create_branch][REASON] Repository is empty. Creating initial commit to enable branching.")
logger.reason("Repository is empty. Creating initial commit to enable branching.", extra={"src": "create_branch"})
readme_path = os.path.join(repo.working_dir, "README.md")
if not os.path.exists(readme_path):
with open(readme_path, "w") as f:
@@ -143,7 +147,7 @@ class GitServiceBranchMixin:
try:
repo.commit(from_branch)
except Exception:
logger.warning(f"[create_branch][REASON] Source branch {from_branch} not found, using HEAD")
logger.reason(f"Source branch {from_branch} not found, using HEAD", extra={"src": "create_branch"})
from_branch = repo.head
try:
new_branch = repo.create_head(name, from_branch)
@@ -160,7 +164,7 @@ class GitServiceBranchMixin:
def checkout_branch(self, dashboard_id: int, name: str):
with belief_scope("GitService.checkout_branch"):
repo = self.get_repo(dashboard_id)
logger.info(f"[checkout_branch][REASON] Checking out branch {name}")
logger.reason(f"Checking out branch {name}", extra={"src": "checkout_branch"})
repo.git.checkout(name)
# [/DEF:checkout_branch:Function]
@@ -174,13 +178,13 @@ class GitServiceBranchMixin:
with belief_scope("GitService.commit_changes"):
repo = self.get_repo(dashboard_id)
if not repo.is_dirty(untracked_files=True) and not files:
logger.info(f"[commit_changes][REASON] No changes to commit for dashboard {dashboard_id}")
logger.reason(f"No changes to commit for dashboard {dashboard_id}", extra={"src": "commit_changes"})
return
if files:
logger.info(f"[commit_changes][REASON] Staging files: {files}")
logger.reason(f"Staging files: {files}", extra={"src": "commit_changes"})
repo.index.add(files)
else:
logger.info("[commit_changes][REASON] Staging all changes")
logger.reason("Staging all changes", extra={"src": "commit_changes"})
repo.git.add(A=True)
repo.index.commit(message)
logger.info(f"[commit_changes][Coherence:OK] Committed changes with message: {message}")

8
backend/src/services/git/_gitea.py
View File

@@ -23,7 +23,7 @@ class GitServiceGiteaMixin:
async def test_connection(self, provider: GitProvider, url: str, pat: str) -> bool:
with belief_scope("GitService.test_connection"):
if ".local" in url or "localhost" in url:
logger.info("[test_connection][REASON] Local/Offline mode detected for URL")
logger.reason("Local/Offline mode detected for URL", extra={"src": "test_connection"})
return True
if not url.startswith(('http://', 'https://')):
logger.error(f"[test_connection][Coherence:Failed] Invalid URL protocol: {url}")
@@ -220,9 +220,9 @@ class GitServiceGiteaMixin:
exc.status_code == 404 and fallback_url and fallback_url != normalized_primary
)
if should_retry_with_fallback:
logger.warning(
"[create_gitea_pull_request][REASON] Primary Gitea URL not found, retrying with remote host: %s",
fallback_url,
logger.explore(
f"Primary Gitea URL not found, retrying with remote host: {fallback_url}",
extra={"src": "create_gitea_pull_request"},
)
active_server_url = fallback_url
try:

2
backend/src/services/git/_status.py
View File

@@ -153,7 +153,7 @@ class GitServiceStatusMixin:
"files_changed": list(commit.stats.files.keys())
})
except Exception as e:
logger.warning(f"[get_commit_history][REASON] Could not retrieve commit history for dashboard {dashboard_id}: {e}")
logger.explore(f"Could not retrieve commit history for dashboard {dashboard_id}: {e}", extra={"src": "get_commit_history"})
return []
return commits
# [/DEF:get_commit_history:Function]

35
backend/src/services/git/_sync.py
View File

@@ -60,9 +60,9 @@ class GitServiceSyncMixin:
finally:
session.close()
except Exception as diag_error:
logger.warning(
"[push_changes][REASON] Failed to load repository binding diagnostics for dashboard %s: %s",
dashboard_id, diag_error,
logger.reason(
"Failed to load repository binding diagnostics",
extra={"src": "push_changes", "dashboard_id": dashboard_id, "error": str(diag_error)},
)
realigned_origin_url = self._align_origin_host_with_config(
dashboard_id=dashboard_id,
@@ -75,13 +75,13 @@ class GitServiceSyncMixin:
origin_urls = list(origin.urls)
except Exception:
origin_urls = []
logger.info(
"[push_changes][REASON] Push diagnostics dashboard=%s config_id=%s config_url=%s binding_remote_url=%s origin_urls=%s origin_realigned=%s",
dashboard_id, binding_config_id, binding_config_url, binding_remote_url, origin_urls, bool(realigned_origin_url),
logger.reason(
f"Push diagnostics dashboard={dashboard_id} config_id={binding_config_id} config_url={binding_config_url} binding_remote_url={binding_remote_url} origin_urls={origin_urls} origin_realigned={bool(realigned_origin_url)}",
extra={"src": "push_changes"},
)
try:
current_branch = repo.active_branch
logger.info(f"[push_changes][REASON] Pushing branch {current_branch.name} to origin")
logger.reason(f"Pushing branch {current_branch.name} to origin", extra={"src": "push_changes"})
tracking_branch = None
try:
tracking_branch = current_branch.tracking_branch()
@@ -120,10 +120,9 @@ class GitServiceSyncMixin:
merge_head_path = os.path.join(repo.git_dir, "MERGE_HEAD")
if os.path.exists(merge_head_path):
payload = self._build_unfinished_merge_payload(repo)
logger.warning(
"[pull_changes][REASON] Unfinished merge detected for dashboard %s (repo_path=%s git_dir=%s branch=%s merge_head=%s merge_msg=%s)",
dashboard_id, payload["repository_path"], payload["git_dir"],
payload["current_branch"], payload["merge_head"], payload["merge_message_preview"],
logger.reason(
f"Unfinished merge detected for dashboard {dashboard_id} (repo_path={payload['repository_path']} git_dir={payload['git_dir']} branch={payload['current_branch']} merge_head={payload['merge_head']} merge_msg={payload['merge_message_preview']})",
extra={"src": "pull_changes"},
)
raise HTTPException(status_code=409, detail=payload)
try:
@@ -133,23 +132,23 @@ class GitServiceSyncMixin:
origin_urls = list(origin.urls)
except Exception:
origin_urls = []
logger.info(
"[pull_changes][REASON] Pull diagnostics dashboard=%s repo_path=%s branch=%s origin_urls=%s",
dashboard_id, repo.working_tree_dir, current_branch, origin_urls,
logger.reason(
f"Pull diagnostics dashboard={dashboard_id} repo_path={repo.working_tree_dir} branch={current_branch} origin_urls={origin_urls}",
extra={"src": "pull_changes"},
)
origin.fetch(prune=True)
remote_ref = f"origin/{current_branch}"
has_remote_branch = any(ref.name == remote_ref for ref in repo.refs)
logger.info(
"[pull_changes][REASON] Pull remote branch check dashboard=%s branch=%s remote_ref=%s exists=%s",
dashboard_id, current_branch, remote_ref, has_remote_branch,
logger.reason(
f"Pull remote branch check dashboard={dashboard_id} branch={current_branch} remote_ref={remote_ref} exists={has_remote_branch}",
extra={"src": "pull_changes"},
)
if not has_remote_branch:
raise HTTPException(
status_code=409,
detail=f"Remote branch '{current_branch}' does not exist yet. Push this branch first.",
)
logger.info(f"[pull_changes][REASON] Pulling changes from origin/{current_branch}")
logger.reason(f"Pulling changes from origin/{current_branch}", extra={"src": "pull_changes"})
repo.git.pull("--no-rebase", "origin", current_branch)
except ValueError:
logger.error(f"[pull_changes][Coherence:Failed] Remote 'origin' not found for dashboard {dashboard_id}")

12
backend/src/services/git/_url.py
View File

@@ -115,9 +115,9 @@ class GitServiceUrlMixin:
aligned_url = self._replace_host_in_url(source_origin_url, config_url)
if not aligned_url:
return None
logger.warning(
"[_align_origin_host_with_config][REASON] Host mismatch for dashboard %s: config_host=%s origin_host=%s, applying origin.set_url",
dashboard_id, config_host, origin_host,
logger.reason(
f"Host mismatch for dashboard {dashboard_id}: config_host={config_host} origin_host={origin_host}, applying origin.set_url",
extra={"src": "_align_origin_host_with_config"},
)
try:
origin.set_url(aligned_url)
@@ -141,9 +141,9 @@ class GitServiceUrlMixin:
finally:
session.close()
except Exception as e:
logger.warning(
"[_align_origin_host_with_config][REASON] Failed to persist aligned remote_url for dashboard %s: %s",
dashboard_id, e,
logger.explore(
f"Failed to persist aligned remote_url for dashboard {dashboard_id}: {e}",
extra={"src": "_align_origin_host_with_config"},
)
return aligned_url
# [/DEF:_align_origin_host_with_config:Function]

8
backend/src/services/notifications/service.py
View File

@@ -113,7 +113,8 @@ class NotificationService:
should_notify = self._should_notify(record, policy)
if not should_notify:
logger.reason(
f"[REASON] Notification skipped for record {record.id} (status={record.status})"
f"Notification skipped for record {record.id} (status={record.status})",
extra={"src": "NotificationService"},
)
return
@@ -127,8 +128,9 @@ class NotificationService:
for channel_type, recipient in targets:
provider = self._providers.get(channel_type)
if not provider:
logger.warning(
f"[NotificationService][EXPLORE] Unsupported or unconfigured channel: {channel_type}"
logger.explore(
f"Unsupported or unconfigured channel: {channel_type}",
extra={"src": "NotificationService"},
)
continue

26
backend/src/services/profile_service.py
View File

@@ -118,7 +118,7 @@ class ProfileService:
with belief_scope(
"ProfileService.get_my_preference", f"user_id={current_user.id}"
):
logger.reflect("[REFLECT] Loading current user's dashboard preference")
logger.reflect("Loading current user's dashboard preference")
preference = self._get_preference_row(current_user.id)
security_summary = self._build_security_summary(current_user)
@@ -189,12 +189,10 @@ class ProfileService:
with belief_scope(
"ProfileService.update_my_preference", f"user_id={current_user.id}"
):
logger.reason(
"[REASON] Evaluating self-scope guard before preference mutation"
)
logger.reason("Evaluating self-scope guard before preference mutation")
requested_user_id = str(target_user_id or current_user.id)
if requested_user_id != str(current_user.id):
logger.explore("[EXPLORE] Cross-user mutation attempt blocked")
logger.explore("Cross-user mutation attempt blocked")
raise ProfileAuthorizationError(
"Cross-user preference mutation is forbidden"
)
@@ -275,7 +273,7 @@ class ProfileService:
email_address=effective_email_address,
)
if validation_errors:
logger.reflect("[REFLECT] Validation failed; mutation is denied")
logger.reflect("Validation failed; mutation is denied")
raise ProfileValidationError(validation_errors)
preference.superset_username = effective_superset_username
@@ -311,7 +309,7 @@ class ProfileService:
preference
)
logger.reason("[REASON] Preference persisted successfully")
logger.reason("Preference persisted successfully")
return ProfilePreferenceResponse(
status="success",
message="Preference saved",
@@ -340,7 +338,7 @@ class ProfileService:
):
environment = self._resolve_environment(request.environment_id)
if environment is None:
logger.explore("[EXPLORE] Lookup aborted: environment not found")
logger.explore("Lookup aborted: environment not found")
raise EnvironmentNotFoundError(
f"Environment '{request.environment_id}' not found"
)
@@ -354,14 +352,14 @@ class ProfileService:
sort_order = "desc"
logger.reflect(
"[REFLECT] Normalized lookup request "
"Normalized lookup request "
f"(env={request.environment_id}, sort_column={sort_column}, sort_order={sort_order}, "
f"page_index={request.page_index}, page_size={request.page_size}, "
f"search={(request.search or '').strip()!r})"
)
try:
logger.reason("[REASON] Performing Superset account lookup")
logger.reason("Performing Superset account lookup")
superset_client = SupersetClient(environment)
adapter = SupersetAccountLookupAdapter(
network_client=superset_client.network,
@@ -389,7 +387,7 @@ class ProfileService:
)
except Exception as exc:
logger.explore(
f"[EXPLORE] Lookup degraded due to upstream error: {exc}"
f"Lookup degraded due to upstream error: {exc}"
)
return SupersetAccountLookupResponse(
status="degraded",
@@ -462,9 +460,9 @@ class ProfileService:
(normalized_resource, normalized_action)
)
except Exception as discovery_error:
logger.warning(
"[ProfileService][EXPLORE] Failed to build declared permission catalog: %s",
discovery_error,
logger.explore(
"Failed to build declared permission catalog",
extra={"error": str(discovery_error), "src": "ProfileService._build_security_summary"},
)
if not declared_permission_pairs:

13
backend/src/services/rbac_permission_catalog.py
View File

@@ -56,10 +56,9 @@ def _discover_route_permissions() -> Set[Tuple[str, str]]:
try:
source = route_file.read_text(encoding="utf-8")
except OSError as read_error:
logger.warning(
"[rbac_permission_catalog][EXPLORE] Failed to read route file %s: %s",
route_file,
read_error,
logger.explore(
"Failed to read route file",
extra={"src": "rbac_permission_catalog", "payload": {"file": str(route_file)}, "error": str(read_error)},
)
continue
@@ -96,9 +95,9 @@ def _discover_plugin_execute_permissions(plugin_loader=None) -> Set[Tuple[str, s
try:
plugin_configs = plugin_loader.get_all_plugin_configs()
except Exception as plugin_error:
logger.warning(
"[rbac_permission_catalog][EXPLORE] Failed to read plugin configs for RBAC discovery: %s",
plugin_error,
logger.explore(
"Failed to read plugin configs for RBAC discovery",
extra={"src": "rbac_permission_catalog", "error": str(plugin_error)},
)
return discovered

2
backend/src/services/resource_service.py
View File

@@ -28,7 +28,7 @@ class ResourceService:
def __init__(self):
with belief_scope("ResourceService.__init__"):
self.git_service = GitService()
logger.info("[ResourceService][REASON] Initialized ResourceService")
logger.reason("Initialized ResourceService", extra={"src": "ResourceService"})
# [/DEF:ResourceService_init:Function]
# [DEF:get_dashboards_with_status:Function]

83
backend/tests/test_smoke_app.py Normal file
View File

@@ -0,0 +1,83 @@
# [DEF:TestSmokeApp:Module]
# @COMPLEXITY: 2
# @SEMANTICS: tests, smoke, app, imports, fastapi
# @PURPOSE: Minimal smoke tests that verify the full application import chain
# succeeds without SyntaxError, IndentationError, ImportError, or
# NameError in any module.
# @LAYER: Tests (Smoke)
# @RELATION: VERIFIES -> src/app.py
# @RELATION: VERIFIES -> src/core/cot_logger.py
# @INVARIANT: All tests must pass without a running PostgreSQL instance.
#
# @RATIONALE: Uses SQLite in-memory database URLs to avoid requiring a live
# PostgreSQL server during smoke testing. The database module's
# _build_engine() already supports SQLite with
# check_same_thread=False.
"""Smoke test: verify the full application imports without errors.
Catches: SyntaxError, IndentationError, ImportError, NameError in any module
that is part of the application import chain.
Database dependency:
The application's database module (src.core.database) creates SQLAlchemy
engines at import time. To avoid needing a running PostgreSQL instance,
environment variables are set to SQLite in-memory databases before the
app is imported.
"""
import os
import sys
from pathlib import Path
# ---------------------------------------------------------------------------
# Database URL override: use SQLite in-memory so no PostgreSQL is needed.
# Must happen before ANY application module is imported.
# ---------------------------------------------------------------------------
os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("TASKS_DATABASE_URL", "sqlite:///:memory:")
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:")
# Ensure the backend root is on sys.path so `src` package is importable
_src_path = str(Path(__file__).resolve().parent.parent)
if _src_path not in sys.path:
sys.path.insert(0, _src_path)
def test_app_imports_cleanly():
"""The application module tree imports without errors."""
from src.app import app
assert app is not None
assert app.title == "Superset Tools API"
def test_app_has_expected_middleware():
"""TraceContextMiddleware is registered (trace_id propagation)."""
from src.app import app
middleware_classes = [m.cls.__name__ for m in app.user_middleware]
assert "TraceContextMiddleware" in middleware_classes, (
f"TraceContextMiddleware not found in app middleware: {middleware_classes}"
)
def test_cot_logger_imports():
"""Core logging infrastructure is importable."""
from src.core.cot_logger import (
log,
seed_trace_id,
get_trace_id,
push_span,
pop_span,
)
tid = seed_trace_id()
assert tid is not None
assert get_trace_id() == tid
log("test", "REASON", "Smoke test log entry", {"ok": True})
# Verify push_span / pop_span round-trip
prev = push_span("test_span")
assert prev == "" # default empty span
pop_span(prev)