#!/usr/bin/env python3 """Live auto-propose E2E for meta-skill-creator. This harness intentionally does not accept a user prompt. It verifies the unattended creator path used by cron and dream hooks: aggregate decision-log history, synthesize a candidate through meta-skill-creator, run its gates, and persist a proposal with auto_* provenance. """ from __future__ import annotations import argparse import asyncio import json import os from datetime import UTC, datetime from pathlib import Path from typing import Any def _load_env_file(path: Path | None) -> None: if path is None or not path.is_file(): return for raw in path.read_text(encoding="utf-8").splitlines(): line = raw.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip().strip("'\"") if key and key not in os.environ: os.environ[key] = value def _seed_history(log_dir: Path) -> Path: log_dir.mkdir(parents=True, exist_ok=True) now = datetime.now(UTC).isoformat() rows: list[dict[str, object]] = [] chains = [ (["history-explorer", "summarize"], 7), (["multi-search-engine", "summarize"], 4), (["weather", "summarize"], 2), ] for chain, count in chains: for _ in range(count): rows.append({ "ts": now, "agent_id": "main", "user_message": "recent decision history operational recap", "skills_invoked": chain, }) path = log_dir / f"decisions-{datetime.now(UTC).strftime('%Y%m%d')}.jsonl" path.write_text( "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows), encoding="utf-8", ) return path async def _run(args: argparse.Namespace) -> dict[str, Any]: home = args.home.expanduser().resolve() log_dir = args.log_dir.expanduser().resolve() if args.log_dir else home / "logs" proposals_dir = ( args.proposals_dir.expanduser().resolve() if args.proposals_dir else home / "proposals" ) workspace_dir = args.workspace.expanduser().resolve() if args.workspace else home / "workspace" home.mkdir(parents=True, exist_ok=True) workspace_dir.mkdir(parents=True, exist_ok=True) if args.seed_history: history_path = _seed_history(log_dir) else: history_path = None os.environ["OPENSQUILLA_STATE_DIR"] = str(home) os.environ["OPENSQUILLA_LOG_DIR"] = str(log_dir) os.environ["OPENSQUILLA_LLM_PROVIDER"] = args.provider os.environ["OPENSQUILLA_LLM_MODEL"] = args.model # Imports happen after env setup so default_opensquilla_home() users resolve # to this isolated state root. from opensquilla.engine.agent import Agent from opensquilla.engine.types import AgentConfig from opensquilla.gateway.boot import _make_auto_propose_tool_context, build_services from opensquilla.gateway.config import GatewayConfig from opensquilla.scheduler.auto_propose_handler import make_auto_propose_handler from opensquilla.scheduler.types import CronJob, SessionTarget from opensquilla.skills.creator.auto_propose import auto_propose from opensquilla.skills.creator.proposer import ( reset_runtime_e2e_context, reset_smoke_fixture_context, set_runtime_e2e_context, set_smoke_fixture_context, ) from opensquilla.skills.creator.runtime_e2e import make_runtime_e2e_context from opensquilla.skills.meta.orchestrator import ( MetaOrchestrator, make_agent_runner_from_parent, make_llm_chat_from_provider, make_tool_invoker_from_handler, ) from opensquilla.tools.dispatch import build_tool_handler text_tiers = { "c0": {"provider": args.provider, "model": args.model, "thinking_level": "off"}, "c1": {"provider": args.provider, "model": args.model, "thinking_level": "low"}, "c2": {"provider": args.provider, "model": args.model, "thinking_level": "medium"}, "c3": {"provider": args.provider, "model": args.model, "thinking_level": "high"}, } actual_cron = args.actual_scheduler and args.trigger == "cron" actual_dream = args.actual_scheduler and args.trigger == "dream" auto_enabled = actual_cron or not args.actual_scheduler config = GatewayConfig( workspace_dir=str(workspace_dir), llm={ "provider": args.provider, "model": args.model, "api_key": os.environ.get("OPENROUTER_API_KEY", ""), "base_url": args.base_url, }, squilla_router={ "enabled": True, "tiers": text_tiers, "default_tier": "c3", }, meta_skill={ "auto_propose": { "enabled": auto_enabled, "cron": args.cron, "on_dream_complete": actual_dream or not args.actual_scheduler, "window_days": args.window_days, "min_freq": args.min_freq, "top_k": args.top_k, "auto_enable": args.auto_enable, "auto_enable_max_risk": args.auto_enable_max_risk, }, }, memory={ "dream": { "enabled": actual_dream, "auto_schedule": actual_dream, "cron": args.cron if actual_dream else None, "preview_mode": True, "min_batch_size": 1, }, }, ) server_handle = None if args.actual_scheduler: from opensquilla.gateway.boot import start_gateway_server server_handle = await start_gateway_server(config=config, run=False) svc = getattr(server_handle, "_services", None) if svc is None: raise RuntimeError("gateway boot did not expose services") else: svc = await build_services( config=config, session_db_path=str(home / "state" / "sessions.sqlite"), seed_agent_workspaces=True, ) assert svc.provider_selector is not None assert svc.tool_registry is not None assert svc.skill_loader is not None def build_orchestrator(agent_id: str) -> MetaOrchestrator: provider_selector = svc.provider_selector clone_selector = getattr(provider_selector, "clone", None) if callable(clone_selector): provider_selector = clone_selector() override_model = getattr(provider_selector, "override_model", None) if callable(override_model): override_model(args.model) provider = provider_selector.resolve() ctx = _make_auto_propose_tool_context( agent_id=agent_id, workspace_dir=str(workspace_dir), ) tool_handler = build_tool_handler(svc.tool_registry, ctx) base_config = AgentConfig( model_id=args.model, workspace_dir=str(workspace_dir), metadata={ "routing_source": "meta_skill_auto_propose", "routing_applied": True, "routed_tier": "c3", "routed_model": args.model, "applied_model": args.model, "thinking_requested": True, "thinking_level": "high", }, ) tool_definitions = svc.tool_registry.to_tool_definitions(ctx) llm_chat = make_llm_chat_from_provider( provider=provider, base_config=base_config, usage_tracker=svc.usage_tracker, session_key=f"auto_propose:{agent_id}", ) base_tool_invoker = make_tool_invoker_from_handler(tool_handler=tool_handler) runtime_e2e_ctx = make_runtime_e2e_context( provider=provider, base_config=base_config, skill_loader=svc.skill_loader, tool_definitions=tool_definitions, tool_handler=tool_handler, agent_factory=Agent, llm_chat=llm_chat, tool_invoker=base_tool_invoker, workspace_dir=str(workspace_dir), usage_tracker=svc.usage_tracker, session_key=f"auto_propose:{agent_id}", tool_registry=svc.tool_registry, tool_context=ctx, system_prompt=base_config.system_prompt or "", baseline_model=args.model, ) async def tool_invoker(tool_name: str, tool_args: dict[str, Any]) -> Any: if tool_name == "meta_skill_persist_proposal": tool_args = dict(tool_args) tool_args.setdefault("home", str(home)) tool_args.setdefault("auto_enable_manual", False) token = set_runtime_e2e_context(runtime_e2e_ctx) smoke_token = set_smoke_fixture_context({"llm_chat": llm_chat}) try: return await base_tool_invoker(tool_name, tool_args) finally: reset_smoke_fixture_context(smoke_token) reset_runtime_e2e_context(token) return MetaOrchestrator( agent_runner=make_agent_runner_from_parent( provider=provider, base_config=base_config, tool_definitions=tool_definitions, tool_handler=tool_handler, agent_factory=Agent, workspace_dir=str(workspace_dir), usage_tracker=svc.usage_tracker, session_key=f"auto_propose:{agent_id}", ), skill_loader=svc.skill_loader, llm_chat=llm_chat, tool_invoker=tool_invoker, workspace_dir=str(workspace_dir), run_writer=getattr(svc, "meta_run_writer", None), triggered_by=f"auto_{args.trigger}", session_key=f"auto_propose:{agent_id}", turn_id=None, usage_tracker=svc.usage_tracker, ) async def scheduler_snapshot() -> dict[str, Any]: scheduler = getattr(svc, "cron_scheduler", None) if scheduler is None: return {"jobs": []} jobs = await scheduler.list_jobs() rows = [] for job in jobs: if not str(getattr(job, "name", "")).startswith(("auto_propose:", "memory_dream:")): continue runs = await scheduler.get_runs(getattr(job, "id"), limit=5) rows.append({ "id": getattr(job, "id", ""), "name": getattr(job, "name", ""), "handler_key": getattr(job, "handler_key", ""), "schedule_kind": str(getattr(job, "schedule_kind", "")), "schedule_raw": getattr(job, "schedule_raw", ""), "status": str(getattr(job, "status", "")), "next_run_at": ( getattr(job, "next_run_at", None).isoformat() if getattr(job, "next_run_at", None) is not None else None ), "run_count": getattr(job, "run_count", 0), "runs": [ { "success": getattr(run, "success", False), "summary": getattr(run, "summary", ""), "delivery_status": getattr(run, "delivery_status", ""), "started_at": ( getattr(run, "started_at", None).isoformat() if getattr(run, "started_at", None) is not None else None ), "finished_at": ( getattr(run, "finished_at", None).isoformat() if getattr(run, "finished_at", None) is not None else None ), } for run in runs ], }) return {"jobs": rows} async def wait_for_automatic_execution() -> dict[str, Any]: deadline = asyncio.get_running_loop().time() + args.wait_seconds last: dict[str, Any] = {} while True: proposals_now = [ sub.name for sub in sorted(proposals_dir.iterdir()) if sub.is_dir() ] if proposals_dir.is_dir() else [] snapshot = await scheduler_snapshot() last = { "triggered_by": args.trigger, "actual_scheduler": True, "proposal_ids": proposals_now, "scheduler": snapshot, } target_prefix = "memory_dream:" if args.trigger == "dream" else "auto_propose:" target_jobs = [ job for job in snapshot.get("jobs", []) if str(job.get("name", "")).startswith(target_prefix) ] target_finished = any( int(job.get("run_count") or 0) > 0 or bool(job.get("runs")) for job in target_jobs ) if target_jobs and target_finished and ( not args.wait_for_proposal or proposals_now ): return last if asyncio.get_running_loop().time() >= deadline: return last await asyncio.sleep(args.poll_seconds) if args.actual_scheduler: result = await wait_for_automatic_execution() elif args.via_handler: handler = make_auto_propose_handler( build_orchestrator=build_orchestrator, skill_loader=svc.skill_loader, log_dir=log_dir, proposals_dir=proposals_dir, config=config.meta_skill.auto_propose, enabled_predicate=lambda: True, ) job = CronJob( id=f"live-auto-propose-{args.trigger}", name="auto_propose:main", cron_expr="* * * * *", schedule_raw="* * * * *", handler_key="auto_propose", payload={"agent_id": "main"}, session_target=SessionTarget.ISOLATED, ) handler_result = await handler(job) result = { "handler": { "summary": handler_result.summary, "delivery_status": handler_result.delivery_status, }, "triggered_by": args.trigger, } else: result_obj = await auto_propose( orchestrator=build_orchestrator("main"), skill_loader=svc.skill_loader, log_dir=log_dir, window_days=args.window_days, min_freq=args.min_freq, top_k=args.top_k, triggered_by=args.trigger, proposals_dir=proposals_dir, auto_enable=args.auto_enable, auto_enable_max_risk=args.auto_enable_max_risk, ) result = { "summary": result_obj.summary(), "proposals_created": result_obj.proposals_created, "proposals_enabled": result_obj.proposals_enabled, "auto_enable": result_obj.auto_enable, "skipped": result_obj.skipped, "errors": result_obj.errors, "triggered_by": result_obj.triggered_by, } proposals = [] if proposals_dir.is_dir(): for sub in sorted(proposals_dir.iterdir()): gates_path = sub / "gates.json" gates = ( json.loads(gates_path.read_text(encoding="utf-8")) if gates_path.is_file() else {} ) proposals.append({ "id": sub.name, "skill": (sub / "SKILL.md").read_text(encoding="utf-8")[:400] if (sub / "SKILL.md").is_file() else "", "gates": gates, }) try: return { "ok": True, "provider": args.provider, "model": args.model, "home": str(home), "log_dir": str(log_dir), "history_path": str(history_path) if history_path else "", "proposals_dir": str(proposals_dir), "result": result, "proposal_count": len(proposals), "proposals": proposals, } finally: if server_handle is not None: await server_handle.close(reason="meta_skill_creator_auto_propose_e2e") else: close = getattr(svc, "close", None) if callable(close): await close() def _parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--env-file", type=Path) parser.add_argument("--home", type=Path, required=True) parser.add_argument("--log-dir", type=Path) parser.add_argument("--proposals-dir", type=Path) parser.add_argument("--workspace", type=Path) parser.add_argument("--provider", default="openrouter") parser.add_argument("--model", default=os.environ.get("OPENROUTER_MODEL", "openai/gpt-4o-mini")) parser.add_argument("--base-url", default="https://openrouter.ai/api/v1") parser.add_argument("--trigger", choices=["cron", "dream"], default="cron") parser.add_argument("--cron", default="* * * * *") parser.add_argument("--via-handler", action="store_true") parser.add_argument("--actual-scheduler", action="store_true") parser.add_argument( "--wait-for-proposal", action="store_true", help=( "For --actual-scheduler, wait until at least one proposal directory " "exists instead of returning as soon as a scheduled run starts." ), ) parser.add_argument("--wait-seconds", type=float, default=120.0) parser.add_argument("--poll-seconds", type=float, default=2.0) parser.add_argument("--seed-history", action="store_true") parser.add_argument("--window-days", type=int, default=30) parser.add_argument("--min-freq", type=int, default=3) parser.add_argument("--top-k", type=int, default=2) parser.add_argument("--auto-enable", action="store_true") parser.add_argument("--auto-enable-max-risk", default="low") return parser def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) _load_env_file(args.env_file) result = asyncio.run(_run(args)) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())