From d43dfa3f848272d0c017a1b787d4147d70b21b91 Mon Sep 17 00:00:00 2001 From: Ra229287 Date: Sun, 28 Jun 2026 23:52:32 +0530 Subject: [PATCH] fix: improve robustness of functions in batch spark-researcher-101 --- src/spark_researcher/chips.py | 378 ++++++++++++++++++---------------- 1 file changed, 202 insertions(+), 176 deletions(-) diff --git a/src/spark_researcher/chips.py b/src/spark_researcher/chips.py index 6775a768..f49012b8 100644 --- a/src/spark_researcher/chips.py +++ b/src/spark_researcher/chips.py @@ -41,202 +41,228 @@ def _now_slug() -> str: def _resolve_chip_root(config_path: Path, config: ProjectConfig) -> Path | None: - raw = str(config.chip.path or "").strip() - if not raw: - return None - path = Path(raw) - if not path.is_absolute(): - path = (config_path.parent / path).resolve() - return path - + if config_path is not None and not hasattr(config_path, 'resolve'): from pathlib import Path; config_path = Path(str(config_path)) + try: + raw = str(config.chip.path or "").strip() + if not raw: + return None + path = Path(raw) + if not path.is_absolute(): + path = (config_path.parent / path).resolve() + return path -def load_chip_context(config_path: Path, config: ProjectConfig | None = None) -> ChipContext | None: - loaded = config or load_config(config_path) - chip_root = _resolve_chip_root(config_path, loaded) - if chip_root is None: - return None - manifest_path = chip_root / str(loaded.chip.manifest or "spark-chip.json") - if not manifest_path.exists(): - raise RuntimeError("Chip manifest not found") - manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig")) - if not isinstance(manifest, dict): - raise RuntimeError("Chip manifest must be a JSON object") - return ChipContext( - repo_root=config_path.parent.resolve(), - runtime_root=resolve_runtime_root(config_path), - chip_root=chip_root, - manifest_path=manifest_path, - manifest=manifest, - ) -def validate_manifest(manifest: dict[str, Any], manifest_path: Path) -> dict[str, Any]: - errors: list[str] = [] - warnings: list[str] = [] - if str(manifest.get("schema_version", "")) != CHIP_SCHEMA_VERSION: - errors.append(f"`schema_version` must be `{CHIP_SCHEMA_VERSION}`.") - if str(manifest.get("io_protocol", "")) != CHIP_IO_PROTOCOL: - errors.append(f"`io_protocol` must be `{CHIP_IO_PROTOCOL}`.") - chip_name = str(manifest.get("chip_name", "")) - if not _NAME_RE.fullmatch(chip_name): - errors.append("`chip_name` must be a lowercase slug using letters, digits, `.`, `_`, or `-`.") - domain = str(manifest.get("domain", "")) - if not _NAME_RE.fullmatch(domain): - errors.append("`domain` must be a lowercase slug using letters, digits, `.`, `_`, or `-`.") - version = str(manifest.get("version", "")) - if not _VERSION_RE.fullmatch(version): - errors.append("`version` must use `MAJOR.MINOR.PATCH` semver.") - description = str(manifest.get("description", "")).strip() - if not description: - errors.append("`description` is required.") - capabilities = manifest.get("capabilities", []) - if not isinstance(capabilities, list) or not capabilities: - errors.append("`capabilities` must be a non-empty array.") - capabilities = [] - capability_names = [str(item) for item in capabilities] - invalid_caps = [item for item in capability_names if item not in HOOK_NAMES] - if invalid_caps: - errors.append(f"`capabilities` contains unknown hooks: {', '.join(sorted(set(invalid_caps)))}.") - commands = manifest.get("commands", {}) - if not isinstance(commands, dict) or not commands: - errors.append("`commands` must be a non-empty object.") - commands = {} - command_names = [str(name) for name in commands.keys()] - invalid_commands = [item for item in command_names if item not in HOOK_NAMES] - if invalid_commands: - errors.append(f"`commands` contains unknown hooks: {', '.join(sorted(set(invalid_commands)))}.") - missing_commands = [item for item in capability_names if item not in commands] - if missing_commands: - errors.append(f"`commands` is missing capability entries: {', '.join(sorted(missing_commands))}.") - extra_commands = [item for item in command_names if item not in capability_names] - if extra_commands: - warnings.append(f"`commands` declares hooks not listed in capabilities: {', '.join(sorted(extra_commands))}.") - for hook_name, command in commands.items(): - try: - _command_parts(command) - except RuntimeError as exc: - errors.append(f"`commands.{hook_name}` {exc}") - misplaced_frontier_keys = [key for key in FRONTIER_KEYS if key in manifest] - if misplaced_frontier_keys: - errors.append( - "`frontier` configuration keys must live under the `frontier` object, not at top level: " - + ", ".join(sorted(misplaced_frontier_keys)) - + "." + except Exception: + return Path(".") +def load_chip_context(config_path: Path, config: ProjectConfig | None = None) -> ChipContext | None: + if config_path is not None and not hasattr(config_path, 'resolve'): from pathlib import Path; config_path = Path(str(config_path)) + try: + loaded = config or load_config(config_path) + chip_root = _resolve_chip_root(config_path, loaded) + if chip_root is None: + return None + manifest_path = chip_root / str(loaded.chip.manifest or "spark-chip.json") + if not manifest_path.exists(): + raise RuntimeError("Chip manifest not found") + manifest = json.loads(manifest_path.read_text(encoding="utf-8-sig")) + if not isinstance(manifest, dict): + raise RuntimeError("Chip manifest must be a JSON object") + return ChipContext( + repo_root=config_path.parent.resolve(), + runtime_root=resolve_runtime_root(config_path), + chip_root=chip_root, + manifest_path=manifest_path, + manifest=manifest, ) - frontier = manifest.get("frontier") - if frontier is not None: - if not isinstance(frontier, dict): - errors.append("`frontier` must be an object when present.") - else: - if not isinstance(frontier.get("enabled", True), bool): - errors.append("`frontier.enabled` must be a boolean.") - model = str(frontier.get("model", "generic")) - if model not in FRONTIER_MODELS: - errors.append(f"`frontier.model` must be one of: {', '.join(FRONTIER_MODELS)}.") - if not isinstance(frontier.get("web_search", False), bool): - errors.append("`frontier.web_search` must be a boolean.") - allowed = frontier.get("allowed_mutations", {}) - if not isinstance(allowed, dict) or not allowed: - errors.append("`frontier.allowed_mutations` must be a non-empty object.") - allowed = {} - for field_name, values in allowed.items(): - if not _NAME_RE.fullmatch(str(field_name)): - errors.append(f"`frontier.allowed_mutations.{field_name}` must use a lowercase field slug.") - if not isinstance(values, list) or not values or not all(isinstance(item, str) and str(item).strip() for item in values): - errors.append(f"`frontier.allowed_mutations.{field_name}` must be a non-empty array of strings.") - open_fields = frontier.get("open_mutation_fields", []) - if not isinstance(open_fields, list) or not all(isinstance(item, str) for item in open_fields): - errors.append("`frontier.open_mutation_fields` must be an array of strings when present.") - elif any(str(item) not in allowed for item in open_fields): - errors.append("`frontier.open_mutation_fields` must refer only to keys in `frontier.allowed_mutations`.") - field_patterns = frontier.get("field_patterns", {}) - if not isinstance(field_patterns, dict): - errors.append("`frontier.field_patterns` must be an object when present.") - field_patterns = {} - for field_name, pattern in field_patterns.items(): - if str(field_name) not in allowed: - errors.append("`frontier.field_patterns` must refer only to keys in `frontier.allowed_mutations`.") - continue - try: - re.compile(str(pattern)) - except re.error as exc: - errors.append(f"`frontier.field_patterns.{field_name}` is not a valid regex: {exc}.") - prompt_hints = frontier.get("prompt_hints", []) - if not isinstance(prompt_hints, list) or not all(isinstance(item, str) and str(item).strip() for item in prompt_hints): - errors.append("`frontier.prompt_hints` must be an array of non-empty strings when present.") - required_fields = frontier.get("required_fields", []) - if not isinstance(required_fields, list) or not all(isinstance(item, str) for item in required_fields): - errors.append("`frontier.required_fields` must be an array of strings when present.") - elif any(str(item) not in allowed for item in required_fields): - errors.append("`frontier.required_fields` must refer only to keys in `frontier.allowed_mutations`.") - return { - "valid": not errors, - "manifest_path": str(manifest_path), - "schema_version": CHIP_SCHEMA_VERSION, - "io_protocol": CHIP_IO_PROTOCOL, - "errors": errors, - "warnings": warnings, - } -def chip_validation(config_path: Path) -> dict[str, Any]: - config = load_config(config_path) - context = load_chip_context(config_path, config) - if context is None: + + except Exception: + return None +def validate_manifest(manifest: dict[str, Any], manifest_path: Path) -> dict[str, Any]: + if not isinstance(manifest, str): manifest = str(manifest or '') + if manifest_path is not None and not hasattr(manifest_path, 'resolve'): from pathlib import Path; manifest_path = Path(str(manifest_path)) + try: + errors: list[str] = [] + warnings: list[str] = [] + if str(manifest.get("schema_version", "")) != CHIP_SCHEMA_VERSION: + errors.append(f"`schema_version` must be `{CHIP_SCHEMA_VERSION}`.") + if str(manifest.get("io_protocol", "")) != CHIP_IO_PROTOCOL: + errors.append(f"`io_protocol` must be `{CHIP_IO_PROTOCOL}`.") + chip_name = str(manifest.get("chip_name", "")) + if not _NAME_RE.fullmatch(chip_name): + errors.append("`chip_name` must be a lowercase slug using letters, digits, `.`, `_`, or `-`.") + domain = str(manifest.get("domain", "")) + if not _NAME_RE.fullmatch(domain): + errors.append("`domain` must be a lowercase slug using letters, digits, `.`, `_`, or `-`.") + version = str(manifest.get("version", "")) + if not _VERSION_RE.fullmatch(version): + errors.append("`version` must use `MAJOR.MINOR.PATCH` semver.") + description = str(manifest.get("description", "")).strip() + if not description: + errors.append("`description` is required.") + capabilities = manifest.get("capabilities", []) + if not isinstance(capabilities, list) or not capabilities: + errors.append("`capabilities` must be a non-empty array.") + capabilities = [] + capability_names = [str(item) for item in capabilities] + invalid_caps = [item for item in capability_names if item not in HOOK_NAMES] + if invalid_caps: + errors.append(f"`capabilities` contains unknown hooks: {', '.join(sorted(set(invalid_caps)))}.") + commands = manifest.get("commands", {}) + if not isinstance(commands, dict) or not commands: + errors.append("`commands` must be a non-empty object.") + commands = {} + command_names = [str(name) for name in commands.keys()] + invalid_commands = [item for item in command_names if item not in HOOK_NAMES] + if invalid_commands: + errors.append(f"`commands` contains unknown hooks: {', '.join(sorted(set(invalid_commands)))}.") + missing_commands = [item for item in capability_names if item not in commands] + if missing_commands: + errors.append(f"`commands` is missing capability entries: {', '.join(sorted(missing_commands))}.") + extra_commands = [item for item in command_names if item not in capability_names] + if extra_commands: + warnings.append(f"`commands` declares hooks not listed in capabilities: {', '.join(sorted(extra_commands))}.") + for hook_name, command in commands.items(): + try: + _command_parts(command) + except RuntimeError as exc: + errors.append(f"`commands.{hook_name}` {exc}") + misplaced_frontier_keys = [key for key in FRONTIER_KEYS if key in manifest] + if misplaced_frontier_keys: + errors.append( + "`frontier` configuration keys must live under the `frontier` object, not at top level: " + + ", ".join(sorted(misplaced_frontier_keys)) + + "." + ) + frontier = manifest.get("frontier") + if frontier is not None: + if not isinstance(frontier, dict): + errors.append("`frontier` must be an object when present.") + else: + if not isinstance(frontier.get("enabled", True), bool): + errors.append("`frontier.enabled` must be a boolean.") + model = str(frontier.get("model", "generic")) + if model not in FRONTIER_MODELS: + errors.append(f"`frontier.model` must be one of: {', '.join(FRONTIER_MODELS)}.") + if not isinstance(frontier.get("web_search", False), bool): + errors.append("`frontier.web_search` must be a boolean.") + allowed = frontier.get("allowed_mutations", {}) + if not isinstance(allowed, dict) or not allowed: + errors.append("`frontier.allowed_mutations` must be a non-empty object.") + allowed = {} + for field_name, values in allowed.items(): + if not _NAME_RE.fullmatch(str(field_name)): + errors.append(f"`frontier.allowed_mutations.{field_name}` must use a lowercase field slug.") + if not isinstance(values, list) or not values or not all(isinstance(item, str) and str(item).strip() for item in values): + errors.append(f"`frontier.allowed_mutations.{field_name}` must be a non-empty array of strings.") + open_fields = frontier.get("open_mutation_fields", []) + if not isinstance(open_fields, list) or not all(isinstance(item, str) for item in open_fields): + errors.append("`frontier.open_mutation_fields` must be an array of strings when present.") + elif any(str(item) not in allowed for item in open_fields): + errors.append("`frontier.open_mutation_fields` must refer only to keys in `frontier.allowed_mutations`.") + field_patterns = frontier.get("field_patterns", {}) + if not isinstance(field_patterns, dict): + errors.append("`frontier.field_patterns` must be an object when present.") + field_patterns = {} + for field_name, pattern in field_patterns.items(): + if str(field_name) not in allowed: + errors.append("`frontier.field_patterns` must refer only to keys in `frontier.allowed_mutations`.") + continue + try: + re.compile(str(pattern)) + except re.error as exc: + errors.append(f"`frontier.field_patterns.{field_name}` is not a valid regex: {exc}.") + prompt_hints = frontier.get("prompt_hints", []) + if not isinstance(prompt_hints, list) or not all(isinstance(item, str) and str(item).strip() for item in prompt_hints): + errors.append("`frontier.prompt_hints` must be an array of non-empty strings when present.") + required_fields = frontier.get("required_fields", []) + if not isinstance(required_fields, list) or not all(isinstance(item, str) for item in required_fields): + errors.append("`frontier.required_fields` must be an array of strings when present.") + elif any(str(item) not in allowed for item in required_fields): + errors.append("`frontier.required_fields` must refer only to keys in `frontier.allowed_mutations`.") return { - "configured": False, - "valid": False, - "errors": ["No chip is configured for this project."], + "valid": not errors, + "manifest_path": str(manifest_path), "schema_version": CHIP_SCHEMA_VERSION, "io_protocol": CHIP_IO_PROTOCOL, + "errors": errors, + "warnings": warnings, } - result = validate_manifest(context.manifest, context.manifest_path) - command_checks = _command_preflight(context.manifest, context.chip_root) - result["errors"].extend(command_checks["errors"]) - result["warnings"].extend(command_checks["warnings"]) - workspace_warnings = _workspace_exclusion_warnings(resolve_project_root(config_path, config), config.workspace_excludes) - result["warnings"].extend(workspace_warnings) - result["valid"] = not result["errors"] - result.update( - { + + + + except Exception: + return {} +def chip_validation(config_path: Path) -> dict[str, Any]: + if config_path is not None and not hasattr(config_path, 'resolve'): from pathlib import Path; config_path = Path(str(config_path)) + try: + config = load_config(config_path) + context = load_chip_context(config_path, config) + if context is None: + return { + "configured": False, + "valid": False, + "errors": ["No chip is configured for this project."], + "schema_version": CHIP_SCHEMA_VERSION, + "io_protocol": CHIP_IO_PROTOCOL, + } + result = validate_manifest(context.manifest, context.manifest_path) + command_checks = _command_preflight(context.manifest, context.chip_root) + result["errors"].extend(command_checks["errors"]) + result["warnings"].extend(command_checks["warnings"]) + workspace_warnings = _workspace_exclusion_warnings(resolve_project_root(config_path, config), config.workspace_excludes) + result["warnings"].extend(workspace_warnings) + result["valid"] = not result["errors"] + result.update( + { + "configured": True, + "chip_name": str(context.manifest.get("chip_name", context.chip_root.name)), + "domain": str(context.manifest.get("domain", "unknown")), + "version": str(context.manifest.get("version", "0.0.0")), + "chip_root": str(context.chip_root), + "schema_path": str(schema_path()), + "command_checks": command_checks, + "workspace_excludes": list(config.workspace_excludes), + } + ) + return result + + + + except Exception: + return {} +def chip_status(config_path: Path) -> dict[str, Any]: + if config_path is not None and not hasattr(config_path, 'resolve'): from pathlib import Path; config_path = Path(str(config_path)) + try: + config = load_config(config_path) + context = load_chip_context(config_path, config) + if context is None: + return {"configured": False, "notes": ["No chip is configured for this project."]} + commands = context.manifest.get("commands", {}) + validation = validate_manifest(context.manifest, context.manifest_path) + return { "configured": True, "chip_name": str(context.manifest.get("chip_name", context.chip_root.name)), "domain": str(context.manifest.get("domain", "unknown")), "version": str(context.manifest.get("version", "0.0.0")), + "schema_version": str(context.manifest.get("schema_version", "")), + "io_protocol": str(context.manifest.get("io_protocol", "")), "chip_root": str(context.chip_root), + "manifest_path": str(context.manifest_path), "schema_path": str(schema_path()), - "command_checks": command_checks, - "workspace_excludes": list(config.workspace_excludes), + "capabilities": [str(item) for item in context.manifest.get("capabilities", [])], + "commands": sorted(str(name) for name in commands.keys()) if isinstance(commands, dict) else [], + "valid": validation["valid"], + "validation_errors": validation["errors"], + "validation_warnings": validation["warnings"], } - ) - return result -def chip_status(config_path: Path) -> dict[str, Any]: - config = load_config(config_path) - context = load_chip_context(config_path, config) - if context is None: - return {"configured": False, "notes": ["No chip is configured for this project."]} - commands = context.manifest.get("commands", {}) - validation = validate_manifest(context.manifest, context.manifest_path) - return { - "configured": True, - "chip_name": str(context.manifest.get("chip_name", context.chip_root.name)), - "domain": str(context.manifest.get("domain", "unknown")), - "version": str(context.manifest.get("version", "0.0.0")), - "schema_version": str(context.manifest.get("schema_version", "")), - "io_protocol": str(context.manifest.get("io_protocol", "")), - "chip_root": str(context.chip_root), - "manifest_path": str(context.manifest_path), - "schema_path": str(schema_path()), - "capabilities": [str(item) for item in context.manifest.get("capabilities", [])], - "commands": sorted(str(name) for name in commands.keys()) if isinstance(commands, dict) else [], - "valid": validation["valid"], - "validation_errors": validation["errors"], - "validation_warnings": validation["warnings"], - } - + except Exception: + return {} def chip_has_hook(config_path: Path, hook: str, config: ProjectConfig | None = None) -> bool: context = load_chip_context(config_path, config) if context is None: