Skills › Scientific Agent Skills › autoskill › run.py
run.py
198 lines · skills/autoskill/scripts/run.py at 5f2bc03
1import datetime as _dt
2from pathlib import Path
4import httpx
6from cluster import cluster_sessions, segment_sessions
7from fetch_window import fetch_window
8from match_skills import load_skill_descriptions, top_k_matches
9from redact import redact
10from synthesize import synthesize
13class ScreenpipeUnreachable(RuntimeError):
14 """Raised when the screenpipe daemon cannot be reached.
16 The autoskill skill cannot run without screenpipe. Install it from
17 https://github.com/screenpipe/screenpipe and start the daemon before
18 invoking this skill.
19 """
22def _default_now() -> str:
23 return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H-%M-%S")
26def _cluster_query(cluster: dict) -> str:
27 parts = ["apps: " + ", ".join(cluster["apps"])]
28 if cluster.get("example_titles"):
29 parts.append("titles: " + "; ".join(cluster["example_titles"]))
30 return " | ".join(parts)
33def _write_plan(proposed_path: Path, clusters: list[dict]) -> None:
34 lines = ["# Dry-run plan", ""]
35 for i, c in enumerate(clusters, 1):
36 lines += [
37 f"## Cluster {i}",
38 f"- apps: {', '.join(c['apps'])}",
39 f"- sessions: {c['session_count']}",
40 f"- total_duration_seconds: {c['total_duration_seconds']}",
41 f"- example titles: {'; '.join(c.get('example_titles', []))}",
42 "",
43 ]
44 (proposed_path / "plan.md").write_text("\n".join(lines))
47def _write_report(proposed_path: Path, results: list[dict]) -> None:
48 lines = ["# autoskill report", ""]
49 if not results:
50 lines.append("No clusters met the minimum size threshold. Nothing to propose.")
51 for r in results:
52 c = r["cluster"]
53 lines += [
54 f"## {', '.join(c['apps'])} — {c['session_count']}× ({c['total_duration_seconds']}s)",
55 f"- verdict: **{r['verdict']}**",
56 ]
57 if r["verdict"] == "reuse":
58 lines.append(f"- matched skill: `{r['target']}`")
59 else:
60 lines.append(f"- draft: `{r['draft_path']}`")
61 lines.append("- top matches:")
62 for s in r["top_k"]:
63 lines.append(f" - `{s['name']}` (score={s['score']:.2f})")
64 lines.append("")
65 (proposed_path / "report.md").write_text("\n".join(lines))
68def run(config, *, start_time, end_time, out_dir,
69 screenpipe_client, backend, embedder, skills_dir,
70 screenpipe_token=None, now=None, dry_run=False):
71 now = now or _default_now
72 try:
73 events = fetch_window(screenpipe_client, start_time, end_time,
74 token=screenpipe_token)
75 except (httpx.ConnectError, httpx.ConnectTimeout) as e:
76 base = getattr(screenpipe_client, "base_url", "http://localhost:3030")
77 raise ScreenpipeUnreachable(
78 f"cannot reach screenpipe at {base}: {e}. "
79 "Install and start the daemon — see "
80 "https://github.com/screenpipe/screenpipe — "
81 "or point config.yaml's screenpipe.url at your instance."
82 ) from e
84 for e in events:
85 e["text"] = redact(e.get("text", ""))
86 e["window_title"] = redact(e.get("window_title", ""))
87 e["app"] = redact(e.get("app", ""))
89 cluster_cfg = config.get("cluster", {})
90 idle_gap = cluster_cfg.get("idle_gap_minutes", 10) * 60
91 min_session = cluster_cfg.get("min_session_minutes", 5) * 60
92 min_cluster = cluster_cfg.get("min_cluster_size", 2)
94 # fetch_window returns ts as ISO strings; convert to epoch for segmentation
95 for e in events:
96 if isinstance(e["ts"], str):
97 e["ts"] = int(_dt.datetime.fromisoformat(e["ts"].replace("Z", "+00:00")).timestamp())
99 sessions = segment_sessions(events, idle_gap_seconds=idle_gap, min_session_seconds=min_session)
100 clusters = cluster_sessions(sessions, min_cluster_size=min_cluster)
102 proposed_path = Path(out_dir) / now()
103 proposed_path.mkdir(parents=True, exist_ok=True)
105 if dry_run:
106 _write_plan(proposed_path, clusters)
107 return proposed_path
109 if not clusters:
110 _write_report(proposed_path, [])
111 return proposed_path
113 skills = load_skill_descriptions(Path(skills_dir))
114 results = []
115 for cluster in clusters:
116 query = _cluster_query(cluster)
117 top_k = top_k_matches(query, skills, embedder=embedder, k=5)
118 decision = synthesize(cluster, top_k, backend=backend)
120 entry = {"cluster": cluster, "top_k": top_k, "verdict": decision["verdict"]}
121 if decision["verdict"] == "reuse":
122 entry["target"] = decision.get("target")
123 else:
124 kind = "new-skills" if decision["verdict"] == "novel" else "composition-recipes"
125 name = decision["name"]
126 draft_dir = proposed_path / kind / name
127 draft_dir.mkdir(parents=True)
128 (draft_dir / "SKILL.md").write_text(decision["skill_body"])
129 entry["draft_path"] = str(draft_dir.relative_to(proposed_path))
130 results.append(entry)
132 _write_report(proposed_path, results)
133 return proposed_path
136def main(argv=None):
137 import argparse
138 import sys
140 import httpx
141 import yaml
143 from backends import check_remote_endpoint, make_backend
145 parser = argparse.ArgumentParser(prog="autoskill")
146 parser.add_argument("--start", required=True, help="ISO start time, e.g. 2026-04-17T00:00:00Z")
147 parser.add_argument("--end", required=True, help="ISO end time")
148 parser.add_argument("--config", default=str(Path(__file__).resolve().parent.parent / "config.yaml"))
149 parser.add_argument("--out", default=None,
150 help="output directory for proposals (default: ~/.autoskill/proposed)")
151 parser.add_argument("--skills-dir", default=None,
152 help="path to skills/ (default: parent of this skill's dir)")
153 parser.add_argument("--dry-run", action="store_true",
154 help="stop after clustering; do not call the LLM backend")
155 args = parser.parse_args(argv)
157 config = yaml.safe_load(Path(args.config).read_text())
159 here = Path(__file__).resolve()
160 skills_dir = Path(args.skills_dir) if args.skills_dir else here.parent.parent.parent
161 out_dir = Path(args.out) if args.out else Path.home() / ".autoskill" / "proposed"
163 import os
164 screenpipe_cfg = config.get("screenpipe", {})
165 screenpipe_url = check_remote_endpoint(
166 screenpipe_cfg.get("url", "http://localhost:3030"), "screenpipe"
167 )
168 screenpipe_token = (screenpipe_cfg.get("token")
169 or os.environ.get("SCREENPIPE_TOKEN"))
170 screenpipe_client = httpx.Client(base_url=screenpipe_url, timeout=60.0)
172 if args.dry_run:
173 backend = None
174 embedder = None
175 else:
176 backend = make_backend(config)
177 from sentence_transformers import SentenceTransformer
178 model = SentenceTransformer(
179 config.get("embeddings", {}).get("model", "sentence-transformers/all-MiniLM-L6-v2"),
180 local_files_only=config.get("embeddings", {}).get("local_files_only", False),
181 )
183 def embedder(text: str):
184 return list(map(float, model.encode(text)))
186 proposed = run(
187 config,
188 start_time=args.start, end_time=args.end, out_dir=out_dir,
189 screenpipe_client=screenpipe_client, backend=backend, embedder=embedder,
190 skills_dir=skills_dir, screenpipe_token=screenpipe_token,
191 dry_run=args.dry_run,
192 )
193 print(f"proposals written to: {proposed}", file=sys.stderr)
194 return 0
197if __name__ == "__main__":
198 raise SystemExit(main())