diff --git a/codewiki/cli/models/job.py b/codewiki/cli/models/job.py index c0c49d12..c3b0ed70 100644 --- a/codewiki/cli/models/job.py +++ b/codewiki/cli/models/job.py @@ -2,7 +2,7 @@ Documentation job data models. """ -from dataclasses import dataclass, field, asdict +from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional, Dict, Any from enum import Enum @@ -44,6 +44,69 @@ class LLMConfig: base_url: str +def _coerce_job_status(value: Any, default: JobStatus = JobStatus.PENDING) -> JobStatus: + """Coerce a raw value into a JobStatus, falling back to a default.""" + if isinstance(value, JobStatus): + return value + if value is None: + return default + try: + return JobStatus(value) + except ValueError: + return default + + +def _coerce_int(value: Any, default: int = 0) -> int: + """Coerce a raw value into an int, falling back to a default.""" + if value is None: + return default + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _coerce_generation_options(value: Any) -> GenerationOptions: + """Coerce a raw dict into a GenerationOptions instance.""" + if isinstance(value, GenerationOptions): + return value + if not value: + return GenerationOptions() + return GenerationOptions( + create_branch=bool(value.get('create_branch', False)), + github_pages=bool(value.get('github_pages', False)), + no_cache=bool(value.get('no_cache', False)), + custom_output=value.get('custom_output'), + ) + + +def _coerce_llm_config(value: Any) -> Optional[LLMConfig]: + """Coerce a raw dict into an LLMConfig instance, or None.""" + if isinstance(value, LLMConfig): + return value + if not value: + return None + return LLMConfig( + main_model=value.get('main_model', ''), + cluster_model=value.get('cluster_model', ''), + base_url=value.get('base_url', ''), + ) + + +def _coerce_statistics(value: Any) -> JobStatistics: + """Coerce a raw dict into a JobStatistics instance.""" + if isinstance(value, JobStatistics): + return value + if not value: + return JobStatistics() + return JobStatistics( + total_files_analyzed=_coerce_int(value.get('total_files_analyzed'), 0), + leaf_nodes=_coerce_int(value.get('leaf_nodes'), 0), + max_depth=_coerce_int(value.get('max_depth'), 0), + total_tokens_used=_coerce_int(value.get('total_tokens_used'), 0), + ) + + @dataclass class DocumentationJob: """ @@ -113,9 +176,23 @@ def to_dict(self) -> Dict[str, Any]: "error_message": self.error_message, "files_generated": self.files_generated, "module_count": self.module_count, - "generation_options": asdict(self.generation_options), - "llm_config": asdict(self.llm_config) if self.llm_config else None, - "statistics": asdict(self.statistics), + "generation_options": { + "create_branch": self.generation_options.create_branch, + "github_pages": self.generation_options.github_pages, + "no_cache": self.generation_options.no_cache, + "custom_output": self.generation_options.custom_output, + }, + "llm_config": { + "main_model": self.llm_config.main_model, + "cluster_model": self.llm_config.cluster_model, + "base_url": self.llm_config.base_url, + } if self.llm_config else None, + "statistics": { + "total_files_analyzed": self.statistics.total_files_analyzed, + "leaf_nodes": self.statistics.leaf_nodes, + "max_depth": self.statistics.max_depth, + "total_tokens_used": self.statistics.total_tokens_used, + }, } return data @@ -135,22 +212,21 @@ def from_dict(cls, data: Dict[str, Any]) -> 'DocumentationJob': branch_name=data.get('branch_name'), timestamp_start=data.get('timestamp_start', datetime.now().isoformat()), timestamp_end=data.get('timestamp_end'), - status=JobStatus(data.get('status', 'pending')), + status=_coerce_job_status(data.get('status'), JobStatus.PENDING), error_message=data.get('error_message'), files_generated=data.get('files_generated', []), - module_count=data.get('module_count', 0), + module_count=_coerce_int(data.get('module_count'), 0), ) # Parse nested objects if 'generation_options' in data: - opts = data['generation_options'] - job.generation_options = GenerationOptions(**opts) + job.generation_options = _coerce_generation_options(data['generation_options']) if 'llm_config' in data and data['llm_config']: - job.llm_config = LLMConfig(**data['llm_config']) + job.llm_config = _coerce_llm_config(data['llm_config']) if 'statistics' in data: - job.statistics = JobStatistics(**data['statistics']) + job.statistics = _coerce_statistics(data['statistics']) return job diff --git a/codewiki/src/config.py b/codewiki/src/config.py index 1e454424..0bd4c906 100644 --- a/codewiki/src/config.py +++ b/codewiki/src/config.py @@ -1,4 +1,4 @@ -from dataclasses import dataclass, field +from dataclasses import dataclass, field, fields, asdict from typing import Optional, List, Dict, Any import argparse import os @@ -42,6 +42,14 @@ def is_cli_context() -> bool: CLUSTER_MODEL = os.getenv('CLUSTER_MODEL', MAIN_MODEL) LLM_BASE_URL = os.getenv('LLM_BASE_URL', 'http://0.0.0.0:4000/') +# Fields that must never be included in serialized output (to_dict()). +# These are runtime-only secrets and should not be persisted, cached, or dumped. +_RUNTIME_ONLY_SECRET_FIELDS = frozenset({ + 'cluster_api_key', + 'main_api_key', + 'fallback_api_key', +}) + @dataclass class Config: """Configuration class for CodeWiki.""" @@ -54,7 +62,7 @@ class Config: main_model: str cluster_model: str fallback_model: str - # Per-provider API keys (required) + # Per-provider API keys (required, runtime-only - never serialized via to_dict()) cluster_api_key: str main_api_key: str fallback_api_key: str @@ -94,6 +102,36 @@ class Config: # When set, all paths are analyzed and merged into unified documentation additional_source_paths: Optional[List[str]] = None + def to_dict(self, include_secrets: bool = False) -> Dict[str, Any]: + """ + Serialize this Config to a plain dict. + + By default, runtime-only secret fields (cluster_api_key, main_api_key, + fallback_api_key) are excluded from the result to prevent accidental + persistence, caching, or logging of API keys. Pass include_secrets=True + only when the caller explicitly needs to reconstruct a fully-functional + Config via from_dict() (e.g. in-process transfer within the same trust + boundary). + """ + data = asdict(self) + if not include_secrets: + for secret_field in _RUNTIME_ONLY_SECRET_FIELDS: + data.pop(secret_field, None) + return data + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> 'Config': + """ + Construct a Config from a dict previously produced by to_dict(). + + If secret fields (cluster_api_key, main_api_key, fallback_api_key) were + excluded (the default for to_dict()), they must be supplied separately + in `data` or this will raise a TypeError due to missing required fields. + """ + known_fields = {f.name for f in fields(cls)} + filtered = {k: v for k, v in data.items() if k in known_fields} + return cls(**filtered) + @property def include_patterns(self) -> Optional[List[str]]: """Get file include patterns from agent instructions.""" @@ -687,4 +725,4 @@ def from_config_manager( agent_instructions=config_obj.agent_instructions.to_dict() if config_obj.agent_instructions else None, diagrams_dir=None, additional_source_paths=additional_paths - ) \ No newline at end of file + )