ourdigital/our-claude-skills
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(reference-curator): implement Python scripts + Gemini quality gate

Browse Source 
Build the refcurator shared Python package and 7 CLI scripts that were
previously specification-only. Add Gemini CLI as an independent pre-distillation
quality evaluator, replacing the circular Claude-self-review pattern.

Key changes:
- shared/lib/src/refcurator/: 7-module package (config, db, models, utils,
  manifest, gemini) with PyMySQL + JSON file dual backend
- 7 Click CLI scripts: discover, crawl_mgr, repo, distiller, reviewer,
  exporter, pipeline — each with subcommands for data management
- Gemini quality gate: evaluates raw content BEFORE distillation using
  5 criteria (relevance, authority, completeness, freshness, distill_value)
- Pipeline reordered: discovery → crawl → store → evaluate → distill → export
- Bug fixes from Codex adversarial review:
  - FileBackend now hard-fails on JOIN/aggregate/GROUP BY queries
  - Exporter uses MAX(review_id) to prevent shipping stale approvals
  - Distiller updates existing rows on refactor instead of forking
- Updated all 7 CLAUDE.md directives with real script references
- install.sh updated with refcurator package install step

51/51 E2E tests passing.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Andrew Yim
2026-04-12 18:19:52 +09:00
parent 133df68b81
commit f215c11c32
23 changed files with 3917 additions and 583 deletions
Download Patch File Download Diff File Expand all files Collapse all files

328
custom-skills/90-reference-curator/07-pipeline-orchestrator/code/CLAUDE.md
View File

@@ -1,6 +1,6 @@
# Pipeline Orchestrator
Coordinates the full 6-skill reference curation workflow with QA loop handling.
Coordinates the full 6-skill reference curation workflow with QA loop handling. Manages pipeline state (init, advance, pause, resume, complete) while Claude orchestrates the actual stage execution.
## Trigger Keywords
"curate references", "full pipeline", "run curation", "reference-curator-pipeline"
@@ -22,257 +22,144 @@ Coordinates the full 6-skill reference curation workflow with QA loop handling.
crawler ─┘
```
## Input Detection
## Pipeline State Management
Parse input to determine mode:
```python
def detect_input_mode(input_value):
if input_value.endswith('.json') and os.path.exists(input_value):
return 'manifest'
elif input_value.startswith('http://') or input_value.startswith('https://'):
return 'urls'
else:
return 'topic'
### Initialize a Run
```bash
uv run python scripts/pipeline.py init \
--input "prompt engineering" --type topic \
--options '{"max_sources": 10, "auto_approve": true}'
```
## Pipeline Execution
### Advance to Next Stage
```bash
uv run python scripts/pipeline.py advance \
--run-id 1 --stage crawling \
--stats '{"sources_discovered": 8}'
```
### Stage 1: Reference Discovery (Topic Mode Only)
### Pause on Error
```bash
uv run python scripts/pipeline.py pause \
--run-id 1 --error "Crawl timeout on page 45" --stage crawling
```
### Resume from Pause
```bash
uv run python scripts/pipeline.py resume --run-id 1
```
### Complete Pipeline
```bash
uv run python scripts/pipeline.py complete \
--run-id 1 --export-path ~/reference-library/exports/ --export-count 40
```
### Check Status
```bash
uv run python scripts/pipeline.py status --run-id 1
uv run python scripts/pipeline.py status --all
```
## QA Loop Tracking
```bash
# Skip if input mode is 'urls' or 'manifest'
if mode == 'topic':
/reference-discovery "$TOPIC" --max-sources $MAX_SOURCES
# Output: manifest.json
# Track a refactor iteration for a document
uv run python scripts/pipeline.py track-iteration \
--run-id 1 --doc-id 42 --action refactor
# Track a deep research iteration
uv run python scripts/pipeline.py track-iteration \
--run-id 1 --doc-id 42 --action deep_research
```
Returns one of:
- `re_distill` — proceed with refactor
- `re_crawl_and_distill` — proceed with deep research
- `needs_manual_review` — max iterations exceeded
| Decision | Max Iterations |
|----------|----------------|
| REFACTOR | 3 |
| DEEP_RESEARCH | 2 |
| Combined total | 5 |
## Pipeline Execution Flow
### Stage 1: Reference Discovery (Topic Mode Only)
```
If mode == 'topic':
→ Claude runs WebSearch
→ discover.py create-manifest
→ discover.py dedup
→ pipeline.py advance --stage crawling
```
### Stage 2: Web Crawler
```bash
# From manifest or URLs
/web-crawler $INPUT --max-pages $MAX_PAGES
# Output: crawled files in ~/reference-library/raw/
```
→ Claude uses Firecrawl MCP tools
→ crawl_mgr.py store-result
→ pipeline.py advance --stage storing
```
### Stage 3: Content Repository
```bash
/content-repository store
# Output: documents stored in MySQL or file-based storage
```
→ repo.py store (for each crawled doc)
→ pipeline.py advance --stage evaluating
```
### Stage 4: Content Distiller
```bash
/content-distiller all-pending
# Output: distilled content records
### Stage 4: Gemini Quality Gate (Pre-Distillation)
```
→ reviewer.py gemini-evaluate-pending --topic "$TOPIC" --auto-approve
→ APPROVE: proceed to distillation
→ DEEP_RESEARCH: pipeline.py track-iteration → crawler (re-crawl)
→ REJECT: skip document entirely
→ pipeline.py advance --stage distilling
```
### Stage 5: Quality Reviewer
```bash
if auto_approve:
/quality-reviewer all-pending --auto-approve --threshold $THRESHOLD
else:
/quality-reviewer all-pending
### Stage 5: Content Distiller (Approved Only)
```
→ distiller.py load-pending
→ Claude distills each approved document
→ distiller.py store
→ pipeline.py advance --stage exporting
```
Handle QA decisions:
- **APPROVE**: Add to export queue
- **REFACTOR**: Re-run distiller with feedback (track iteration count)
- **DEEP_RESEARCH**: Run crawler for additional sources, then distill
- **REJECT**: Archive with reason
### Stage 6: Markdown Exporter
```bash
/markdown-exporter $EXPORT_FORMAT
# Output: files in ~/reference-library/exports/
```
## State Management
### Initialize Pipeline State
```python
def init_pipeline_state(run_id, input_value, options):
state = {
"run_id": run_id,
"run_type": detect_input_mode(input_value),
"input_value": input_value,
"status": "running",
"current_stage": "discovery",
"options": options,
"stats": {
"sources_discovered": 0,
"pages_crawled": 0,
"documents_stored": 0,
"documents_distilled": 0,
"approved": 0,
"refactored": 0,
"deep_researched": 0,
"rejected": 0,
"needs_manual_review": 0
},
"started_at": datetime.now().isoformat()
}
save_state(run_id, state)
return state
```
### MySQL State (Preferred)
```sql
INSERT INTO pipeline_runs (run_type, input_value, options)
VALUES ('topic', 'Claude system prompts', '{"max_sources": 10}');
```
### File-Based Fallback
```
~/reference-library/pipeline_state/run_XXX/
├── state.json # Current stage and stats
├── manifest.json # Discovered sources
├── crawl_results.json # Crawled document paths
├── review_log.json # QA decisions per document
└── errors.log # Any errors encountered
```
## QA Loop Logic
```python
MAX_REFACTOR_ITERATIONS = 3
MAX_DEEP_RESEARCH_ITERATIONS = 2
MAX_TOTAL_ITERATIONS = 5
def handle_qa_decision(doc_id, decision, iteration_counts):
refactor_count = iteration_counts.get('refactor', 0)
research_count = iteration_counts.get('deep_research', 0)
total = refactor_count + research_count
if total >= MAX_TOTAL_ITERATIONS:
return 'needs_manual_review'
if decision == 'refactor':
if refactor_count >= MAX_REFACTOR_ITERATIONS:
return 'needs_manual_review'
iteration_counts['refactor'] = refactor_count + 1
return 're_distill'
if decision == 'deep_research':
if research_count >= MAX_DEEP_RESEARCH_ITERATIONS:
return 'needs_manual_review'
iteration_counts['deep_research'] = research_count + 1
return 're_crawl_and_distill'
return decision # approve or reject
→ exporter.py project
→ exporter.py index
→ exporter.py crossrefs
→ exporter.py verify
→ pipeline.py complete
```
## Checkpoint Strategy
Save checkpoint after each stage completes:
| Stage | Checkpoint | Resume Point |
|-------|------------|--------------|
| discovery | `manifest.json` created | → crawler |
| crawl | `crawl_results.json` | → repository |
| store | DB records or file list | → distiller |
| discovery | manifest.json created | → crawler |
| crawl | crawl_result.json | → repository |
| store | DB records | → distiller |
| distill | distilled_content records | → reviewer |
| review | review_logs records | → exporter or loop |
| export | final export complete | Done |
## Progress Reporting
## Scripts
Report progress to user at key checkpoints:
```
[Pipeline] Stage 1/6: Discovery - Found 8 sources
[Pipeline] Stage 2/6: Crawling - 45/50 pages complete
[Pipeline] Stage 3/6: Storing - 45 documents saved
[Pipeline] Stage 4/6: Distilling - 45 documents processed
[Pipeline] Stage 5/6: Reviewing - 40 approved, 3 refactored, 2 rejected
[Pipeline] Stage 6/6: Exporting - 40 documents exported
[Pipeline] Complete! See ~/reference-library/exports/
```
## Error Handling
```python
def handle_stage_error(stage, error, state):
state['status'] = 'paused'
state['error_message'] = str(error)
state['error_stage'] = stage
save_state(state['run_id'], state)
# Log to errors.log
log_error(state['run_id'], stage, error)
# Report to user
return f"Pipeline paused at {stage}: {error}. Resume with run_id {state['run_id']}"
```
## Resume Pipeline
```python
def resume_pipeline(run_id):
state = load_state(run_id)
if state['status'] != 'paused':
return f"Pipeline {run_id} is {state['status']}, cannot resume"
stage = state['current_stage']
state['status'] = 'running'
state['error_message'] = None
save_state(run_id, state)
# Resume from failed stage
return execute_from_stage(stage, state)
```
## Output Summary
On completion, generate summary:
```json
{
"run_id": 123,
"status": "completed",
"duration_minutes": 15,
"stats": {
"sources_discovered": 5,
"pages_crawled": 45,
"documents_stored": 45,
"documents_distilled": 45,
"approved": 40,
"refactored": 8,
"deep_researched": 2,
"rejected": 3,
"needs_manual_review": 2
},
"exports": {
"format": "project_files",
"path": "~/reference-library/exports/",
"document_count": 40
},
"errors": []
}
```
## Integration Points
| Skill | Called By | Provides |
|-------|-----------|----------|
| reference-discovery | Orchestrator | manifest.json |
| web-crawler | Orchestrator | Raw crawled files |
| content-repository | Orchestrator | Stored documents |
| content-distiller | Orchestrator, QA loop | Distilled content |
| quality-reviewer | Orchestrator | QA decisions |
| markdown-exporter | Orchestrator | Final exports |
| Command | Purpose |
|---------|---------|
| `pipeline.py init` | Initialize a new pipeline run |
| `pipeline.py advance` | Advance to next stage |
| `pipeline.py pause` | Pause on error |
| `pipeline.py resume` | Resume from pause |
| `pipeline.py complete` | Mark pipeline complete |
| `pipeline.py status` | Show run status |
| `pipeline.py track-iteration` | Track QA loop iterations |
## Configuration
Read from `~/.config/reference-curator/pipeline_config.yaml`:
Reads from `~/.config/reference-curator/pipeline_config.yaml`:
```yaml
pipeline:
@@ -288,9 +175,4 @@ qa_loop:
export:
default_format: project_files
include_rejected: false
state:
backend: mysql # or 'file'
state_directory: ~/reference-library/pipeline_state/
```

339
custom-skills/90-reference-curator/07-pipeline-orchestrator/code/scripts/pipeline.py Normal file
View File

@@ -0,0 +1,339 @@
#!/usr/bin/env python3
"""Pipeline Orchestrator CLI — manage pipeline runs and state.
This script provides state management for the 6-stage pipeline.
Claude orchestrates the actual stages via slash commands.
Usage:
python pipeline.py init --input "prompt engineering" --type topic [--options '{"max_sources": 10}']
python pipeline.py advance --run-id 1 --stage crawling [--stats '{"pages_crawled": 45}']
python pipeline.py pause --run-id 1 --error "Crawl timeout" --stage crawling
python pipeline.py resume --run-id 1
python pipeline.py complete --run-id 1 [--export-path ~/reference-library/exports/]
python pipeline.py status [--run-id 1]
python pipeline.py track-iteration --run-id 1 --doc-id 42 --action refactor
"""
import json
import sys
from datetime import datetime
from pathlib import Path
import click
from rich.console import Console
from rich.table import Table
from refcurator.db import db_session
from refcurator.config import get_pipeline_config
console = Console()
# QA loop limits
MAX_REFACTOR = 3
MAX_DEEP_RESEARCH = 2
MAX_TOTAL = 5
@click.group()
def cli():
"""Pipeline Orchestrator — manage pipeline run state."""
pass
@cli.command()
@click.option("--input", "input_value", required=True, help="Topic, URL(s), or manifest path")
@click.option("--type", "run_type", required=True,
type=click.Choice(["topic", "urls", "manifest"]))
@click.option("--options", type=str, help="Pipeline options as JSON string")
def init(input_value, run_type, options):
"""Initialize a new pipeline run."""
opts = json.loads(options) if options else {}
# Merge with config defaults
try:
cfg = get_pipeline_config()
defaults = cfg.get("pipeline", {})
for key in ("max_sources", "max_pages", "auto_approve", "approval_threshold"):
if key not in opts and key in defaults:
opts[key] = defaults[key]
except FileNotFoundError:
pass
stats = {
"sources_discovered": 0,
"pages_crawled": 0,
"documents_stored": 0,
"documents_distilled": 0,
"approved": 0,
"refactored": 0,
"deep_researched": 0,
"rejected": 0,
"needs_manual_review": 0,
}
# Determine starting stage
start_stage = "discovery" if run_type == "topic" else "crawling"
with db_session() as db:
run_id = db.insert_returning_id(
"""INSERT INTO pipeline_runs
(run_type, input_value, status, current_stage, options, stats, started_at)
VALUES (%s, %s, %s, %s, %s, %s, %s)""",
(run_type, input_value, "running", start_stage,
json.dumps(opts), json.dumps(stats), datetime.now().isoformat()),
)
console.print(f"[green]Pipeline initialized:[/green] run_id={run_id}")
console.print(f" Type: {run_type}")
console.print(f" Input: {input_value}")
console.print(f" Starting stage: {start_stage}")
click.echo(json.dumps({"run_id": run_id, "status": "running", "stage": start_stage}))
@cli.command()
@click.option("--run-id", required=True, type=int, help="Pipeline run ID")
@click.option("--stage", required=True,
type=click.Choice(["discovery", "crawling", "storing", "evaluating",
"distilling", "exporting"]))
@click.option("--stats", type=str, help="Stats update as JSON string (merged with existing)")
def advance(run_id, stage, stats):
"""Advance pipeline to the next stage."""
stats_update = json.loads(stats) if stats else {}
with db_session() as db:
run = db.fetch_one(
"SELECT * FROM pipeline_runs WHERE run_id = %s", (run_id,)
)
if not run:
console.print(f"[red]Error:[/red] run_id={run_id} not found")
sys.exit(1)
if run["status"] != "running":
console.print(f"[red]Error:[/red] Pipeline is {run['status']}, not running")
sys.exit(1)
# Merge stats
current_stats = json.loads(run["stats"]) if isinstance(run["stats"], str) else (run["stats"] or {})
current_stats.update(stats_update)
db.execute(
"""UPDATE pipeline_runs
SET current_stage = %s, stats = %s
WHERE run_id = %s""",
(stage, json.dumps(current_stats), run_id),
)
stage_num = ["discovery", "crawling", "storing", "evaluating", "distilling", "exporting"].index(stage) + 1
console.print(f"[green]Pipeline advanced:[/green] Stage {stage_num}/6 — {stage}")
click.echo(json.dumps({"run_id": run_id, "stage": stage, "stats": current_stats}))
@cli.command()
@click.option("--run-id", required=True, type=int, help="Pipeline run ID")
@click.option("--error", required=True, help="Error message")
@click.option("--stage", required=True, help="Stage where error occurred")
def pause(run_id, error, stage):
"""Pause pipeline due to error."""
with db_session() as db:
db.execute(
"""UPDATE pipeline_runs
SET status = %s, error_message = %s, error_stage = %s
WHERE run_id = %s""",
("paused", error, stage, run_id),
)
console.print(f"[yellow]Pipeline paused:[/yellow] run_id={run_id} at {stage}")
console.print(f" Error: {error}")
console.print(f" Resume with: python pipeline.py resume --run-id {run_id}")
@cli.command()
@click.option("--run-id", required=True, type=int, help="Pipeline run ID to resume")
def resume(run_id):
"""Resume a paused pipeline."""
with db_session() as db:
run = db.fetch_one(
"SELECT * FROM pipeline_runs WHERE run_id = %s", (run_id,)
)
if not run:
console.print(f"[red]Error:[/red] run_id={run_id} not found")
sys.exit(1)
if run["status"] != "paused":
console.print(f"[red]Error:[/red] Pipeline is {run['status']}, not paused")
sys.exit(1)
db.execute(
"""UPDATE pipeline_runs
SET status = %s, error_message = %s
WHERE run_id = %s""",
("running", None, run_id),
)
stage = run["current_stage"]
console.print(f"[green]Pipeline resumed:[/green] run_id={run_id}, continuing from {stage}")
click.echo(json.dumps({"run_id": run_id, "status": "running", "resume_stage": stage}))
@cli.command()
@click.option("--run-id", required=True, type=int, help="Pipeline run ID")
@click.option("--export-path", type=click.Path(), help="Export output path")
@click.option("--export-count", type=int, help="Number of documents exported")
def complete(run_id, export_path, export_count):
"""Mark pipeline as completed."""
with db_session() as db:
db.execute(
"""UPDATE pipeline_runs
SET status = %s, completed_at = %s, export_path = %s, export_document_count = %s
WHERE run_id = %s""",
("completed", datetime.now().isoformat(), export_path, export_count, run_id),
)
run = db.fetch_one("SELECT * FROM pipeline_runs WHERE run_id = %s", (run_id,))
stats = json.loads(run["stats"]) if isinstance(run["stats"], str) else (run["stats"] or {})
console.print(f"\n[bold green]Pipeline Complete![/bold green] run_id={run_id}")
console.print(f" Sources: {stats.get('sources_discovered', 0)}")
console.print(f" Crawled: {stats.get('pages_crawled', 0)}")
console.print(f" Stored: {stats.get('documents_stored', 0)}")
console.print(f" Approved: {stats.get('approved', 0)}")
console.print(f" Rejected: {stats.get('rejected', 0)}")
if export_path:
console.print(f" Exports: {export_path}")
click.echo(json.dumps({"run_id": run_id, "status": "completed", "stats": stats}, default=str))
@cli.command()
@click.option("--run-id", type=int, help="Show specific run (or latest if omitted)")
@click.option("--all", "show_all", is_flag=True, help="Show all runs")
def status(run_id, show_all):
"""Show pipeline run status."""
with db_session() as db:
if run_id:
rows = db.fetch_all(
"SELECT * FROM pipeline_runs WHERE run_id = %s", (run_id,)
)
elif show_all:
rows = db.fetch_all(
"SELECT * FROM pipeline_runs ORDER BY started_at DESC LIMIT %s", (20,)
)
else:
rows = db.fetch_all(
"SELECT * FROM pipeline_runs ORDER BY started_at DESC LIMIT %s", (1,)
)
if not rows:
console.print("[dim]No pipeline runs found.[/dim]")
return
for run in rows:
stats = json.loads(run["stats"]) if isinstance(run["stats"], str) else (run["stats"] or {})
status_color = {
"running": "green", "completed": "blue",
"failed": "red", "paused": "yellow",
}.get(run["status"], "white")
console.print(f"\n[bold]Pipeline Run #{run['run_id']}[/bold]")
console.print(f" Status: [{status_color}]{run['status']}[/{status_color}]")
console.print(f" Type: {run['run_type']}")
console.print(f" Input: {run['input_value']}")
console.print(f" Stage: {run['current_stage']}")
console.print(f" Started: {run['started_at']}")
if run.get("error_message"):
console.print(f" [red]Error:[/red] {run['error_message']} (at {run.get('error_stage')})")
if stats:
table = Table(title="Pipeline Stats", show_header=False)
table.add_column("Metric")
table.add_column("Value", justify="right")
for k, v in stats.items():
if v:
table.add_row(k.replace("_", " ").title(), str(v))
console.print(table)
@cli.command("track-iteration")
@click.option("--run-id", required=True, type=int, help="Pipeline run ID")
@click.option("--doc-id", required=True, type=int, help="Document ID")
@click.option("--action", required=True, type=click.Choice(["refactor", "deep_research"]),
help="QA loop action")
def track_iteration(run_id, doc_id, action):
"""Track QA loop iteration for a document.
Returns the routing decision:
- 're_distill': proceed with refactor
- 're_crawl_and_distill': proceed with deep research
- 'needs_manual_review': max iterations exceeded
"""
with db_session() as db:
tracker = db.fetch_one(
"""SELECT * FROM pipeline_iteration_tracker
WHERE run_id = %s AND doc_id = %s""",
(run_id, doc_id),
)
if tracker:
refactor_count = tracker.get("refactor_count", 0)
deep_research_count = tracker.get("deep_research_count", 0)
else:
refactor_count = 0
deep_research_count = 0
total = refactor_count + deep_research_count
# Check limits
if total >= MAX_TOTAL:
result = "needs_manual_review"
elif action == "refactor" and refactor_count >= MAX_REFACTOR:
result = "needs_manual_review"
elif action == "deep_research" and deep_research_count >= MAX_DEEP_RESEARCH:
result = "needs_manual_review"
elif action == "refactor":
refactor_count += 1
result = "re_distill"
else:
deep_research_count += 1
result = "re_crawl_and_distill"
# Upsert tracker
if tracker:
db.execute(
"""UPDATE pipeline_iteration_tracker
SET refactor_count = %s, deep_research_count = %s,
final_decision = %s
WHERE run_id = %s AND doc_id = %s""",
(refactor_count, deep_research_count,
"needs_manual_review" if result == "needs_manual_review" else None,
run_id, doc_id),
)
else:
db.insert_returning_id(
"""INSERT INTO pipeline_iteration_tracker
(run_id, doc_id, refactor_count, deep_research_count, final_decision)
VALUES (%s, %s, %s, %s, %s)""",
(run_id, doc_id, refactor_count, deep_research_count,
"needs_manual_review" if result == "needs_manual_review" else None),
)
console.print(
f"[{'yellow' if result == 'needs_manual_review' else 'green'}]"
f"Iteration tracked:[/] doc_id={doc_id}, {action} "
f"(refactor: {refactor_count}/{MAX_REFACTOR}, "
f"research: {deep_research_count}/{MAX_DEEP_RESEARCH}) → {result}"
)
click.echo(json.dumps({
"run_id": run_id,
"doc_id": doc_id,
"action": action,
"result": result,
"refactor_count": refactor_count,
"deep_research_count": deep_research_count,
"total_iterations": refactor_count + deep_research_count,
}))
if __name__ == "__main__":
cli()