""" ThinkStorm Processing Pipeline & Orchestration Engine Implements bounded processors, untrusted content safety boundaries, provenance logging, and token accounting. """ import re import json import time import asyncio from typing import Dict, Any, List, Optional, Tuple from ..database import get_db, next_sequence, get_utc_now from ..models import ( Idea, IdeaURL, LifecycleState, ProcessingState, URLSafetyState, AutomationPolicy, ProcessorStatus, ProcessorRun, WorkTrackState ) from ..prompts.catalog import get_prompt_version, select_aligned_profile from ..services.omniroute import OmniRouteAdapter from ..services.searxng import SearXNGAdapter from ..services.perplexica import PerplexicaAdapter from ..services.opengist import OpenGistAdapter from ..services.gitea import GiteaAdapter from ..services.virustotal import VirusTotalAdapter omniroute_svc = OmniRouteAdapter() searxng_svc = SearXNGAdapter() perplexica_svc = PerplexicaAdapter() opengist_svc = OpenGistAdapter() gitea_svc = GiteaAdapter() virustotal_svc = VirusTotalAdapter() URL_REGEX = re.compile(r'https?://[^\s<>"\']+', re.IGNORECASE) async def record_processor_run( idea_id: str, processor_name: str, stage: str, prompt_id: Optional[str], prompt_version: Optional[int], prompt_hash: Optional[str], model_policy: str, resolved_provider: str, resolved_model: str, input_tokens: int, output_tokens: int, total_tokens: int, started_at: str, completed_at: str, duration_ms: int, output_artifact: Optional[str], output_data: Dict[str, Any], status: ProcessorStatus = ProcessorStatus.COMPLETED, error_message: Optional[str] = None, work_track_id: Optional[str] = None ) -> str: """Creates an immutable provenance execution record.""" run_id = next_sequence("run") with get_db() as conn: conn.execute( """ INSERT INTO processor_runs ( id, idea_id, work_track_id, processor_name, stage, prompt_id, prompt_version, prompt_hash, model_policy, resolved_provider, resolved_model, input_tokens, output_tokens, total_tokens, started_at, completed_at, duration_ms, output_artifact, output_data, error_message, status ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( run_id, idea_id, work_track_id, processor_name, stage, prompt_id, prompt_version, prompt_hash, model_policy, resolved_provider, resolved_model, input_tokens, output_tokens, total_tokens, started_at, completed_at, duration_ms, output_artifact, json.dumps(output_data), error_message, status.value ) ) return run_id # ------------------------------------------------------------- # Processor 1: URL Extraction & Safety Assessment # ------------------------------------------------------------- async def process_url_safety(idea_id: str, submission_text: str) -> Tuple[List[IdeaURL], bool]: """ Extracts URLs from submission text and applies VirusTotal safety policy. Rule: Any VirusTotal malicious detection (malicious > 0) blocks automated retrieval and quarantines idea. """ extracted_urls = URL_REGEX.findall(submission_text) # Deduplicate and bound unique_urls = list(dict.fromkeys(extracted_urls))[:10] url_records: List[IdeaURL] = [] quarantine_required = False with get_db() as conn: for url_str in unique_urls: assessment = await virustotal_svc.assess_url(url_str) safety_state = URLSafetyState(assessment["safety_state"]) policy = AutomationPolicy(assessment["automation_policy"]) vt_data = assessment.get("virustotal", {}) admin_req = assessment.get("quarantine_required", False) if admin_req: quarantine_required = True cursor = conn.execute( """ INSERT INTO idea_urls (idea_id, url, safety_state, automation_policy, virustotal_data, admin_review_required) VALUES (?, ?, ?, ?, ?, ?) """, (idea_id, url_str, safety_state.value, policy.value, json.dumps(vt_data), 1 if admin_req else 0) ) url_records.append(IdeaURL( id=cursor.lastrowid, idea_id=idea_id, url=url_str, safety_state=safety_state, automation_policy=policy, virustotal_data=vt_data, admin_review_required=admin_req )) return url_records, quarantine_required # ------------------------------------------------------------- # Processor 2: Normalization & Classification # ------------------------------------------------------------- async def process_normalization(idea_id: str, original_text: str) -> Dict[str, Any]: """Generates Title, Summary, Categories, Tags, and assigns Prompt Profile.""" start_time = get_utc_now() prompt_info = get_prompt_version("normalize-idea", is_admin=True) if not prompt_info: raise ValueError("Prompt 'normalize-idea' not found.") system_prompt = prompt_info["system_prompt"] user_prompt = prompt_info["user_prompt_template"].replace("{{submission_text}}", original_text) llm_resp = await omniroute_svc.chat_completion( system_prompt=system_prompt, user_prompt=user_prompt, model_policy="fast", max_tokens=600 ) parsed = omniroute_svc.extract_json(llm_resp["text"]) title = parsed.get("title", "").strip() if not title or title.lower() in ("untitled idea", "untitled", "new idea", "null"): # Synthesize a smart title from first sentence first_sentence = original_text.split(".")[0].replace("\n", " ").strip() words = [w for w in first_sentence.split() if not w.startswith("http")][:8] title = " ".join(words).title() or "Self-Hosted Platform Project" if len(title) > 60: title = title[:57] + "..." summary = parsed.get("summary", "").strip() if not summary or len(summary) < 20: summary = original_text.strip() if len(summary) > 300: summary = summary[:297] + "..." categories = parsed.get("categories", ["Software Development"]) tags = parsed.get("tags", ["self-hosted", "automation"]) # Align prompt profile profile_id = parsed.get("suggested_profile") or select_aligned_profile(categories, tags) end_time = get_utc_now() await record_processor_run( idea_id=idea_id, processor_name="NormalizeIdea", stage="NORMALIZATION", prompt_id="normalize-idea", prompt_version=prompt_info["version"], prompt_hash=prompt_info["prompt_hash"], model_policy="fast", resolved_provider=llm_resp["resolved_provider"], resolved_model=llm_resp["resolved_model"], input_tokens=llm_resp["input_tokens"], output_tokens=llm_resp["output_tokens"], total_tokens=llm_resp["total_tokens"], started_at=start_time, completed_at=end_time, duration_ms=llm_resp["duration_ms"], output_artifact="metadata.json", output_data=parsed ) # Persist normalized data to DB with get_db() as conn: conn.execute( """ UPDATE ideas SET title = ?, summary = ?, profile_id = ?, enrichment_level = 1, updated_at = ? WHERE id = ? """, (title, summary, profile_id, end_time, idea_id) ) # Store categories for cat in categories: conn.execute("INSERT OR IGNORE INTO categories (name) VALUES (?)", (cat,)) c_row = conn.execute("SELECT id FROM categories WHERE name = ?", (cat,)).fetchone() if c_row: conn.execute("INSERT OR IGNORE INTO idea_categories (idea_id, category_id) VALUES (?, ?)", (idea_id, c_row["id"])) # Store tags for t in tags: clean_tag = t.lstrip("#").lower().strip() conn.execute("INSERT OR IGNORE INTO tags (name) VALUES (?)", (clean_tag,)) t_row = conn.execute("SELECT id FROM tags WHERE name = ?", (clean_tag,)).fetchone() if t_row: conn.execute("INSERT OR IGNORE INTO idea_tags (idea_id, tag_id) VALUES (?, ?)", (idea_id, t_row["id"])) return { "title": title, "summary": summary, "categories": categories, "tags": tags, "profile_id": profile_id } # ------------------------------------------------------------- # Processor 3: Duplicate & Relationship Detector # ------------------------------------------------------------- async def process_duplicate_detection(idea_id: str, title: str, summary: str) -> Dict[str, Any]: start_time = get_utc_now() prompt_info = get_prompt_version("duplicate-check", is_admin=True) # Fetch recent ideas summary with get_db() as conn: rows = conn.execute("SELECT id, title, summary FROM ideas WHERE id != ? ORDER BY submitted_at DESC LIMIT 15", (idea_id,)).fetchall() catalog_summary = "\n".join([f"- {r['id']}: {r['title']} - {r['summary'][:80]}..." for r in rows]) or "No previous ideas in catalog." user_prompt = prompt_info["user_prompt_template"].replace("{{title}}", title).replace("{{summary}}", summary).replace("{{catalog_summary}}", catalog_summary) llm_resp = await omniroute_svc.chat_completion( system_prompt=prompt_info["system_prompt"], user_prompt=user_prompt, model_policy="fast", max_tokens=400 ) parsed = omniroute_svc.extract_json(llm_resp["text"]) is_dup = parsed.get("is_duplicate", False) target_id = parsed.get("duplicate_target_id") related_ids = parsed.get("related_ids", []) rationale = parsed.get("rationale", "") end_time = get_utc_now() await record_processor_run( idea_id=idea_id, processor_name="DetectDuplicates", stage="DUPLICATE_CHECK", prompt_id="duplicate-check", prompt_version=prompt_info["version"], prompt_hash=prompt_info["prompt_hash"], model_policy="fast", resolved_provider=llm_resp["resolved_provider"], resolved_model=llm_resp["resolved_model"], input_tokens=llm_resp["input_tokens"], output_tokens=llm_resp["output_tokens"], total_tokens=llm_resp["total_tokens"], started_at=start_time, completed_at=end_time, duration_ms=llm_resp["duration_ms"], output_artifact=None, output_data=parsed ) with get_db() as conn: if is_dup and target_id: conn.execute( "INSERT INTO idea_relationships (source_idea_id, target_idea_id, relationship_type, notes, created_at) VALUES (?, ?, 'DUPLICATES', ?, ?)", (idea_id, target_id, rationale, end_time) ) conn.execute("UPDATE ideas SET lifecycle_state = 'DUPLICATE' WHERE id = ?", (idea_id,)) for rel_id in related_ids: if rel_id != idea_id: conn.execute( "INSERT INTO idea_relationships (source_idea_id, target_idea_id, relationship_type, notes, created_at) VALUES (?, ?, 'RELATED', ?, ?)", (idea_id, rel_id, rationale, end_time) ) return parsed # ------------------------------------------------------------- # Processor 4: Prior Art & Web Discovery (SearXNG) # ------------------------------------------------------------- async def process_prior_art(idea_id: str, title: str, summary: str, original_text: str, tags: List[str]) -> str: start_time = get_utc_now() # 1. Search SearXNG search_query = f"{title} {' '.join(tags[:3])} open source software alternative" search_results = await searxng_svc.search(search_query, limit=6) formatted_results = "\n\n".join([ f"[{i+1}] {r['title']} ({r['url']})\n{r['content']}" for i, r in enumerate(search_results) ]) or "No relevant search results found." # 2. Synthesize via LLM prompt_info = get_prompt_version("prior-art-search", is_admin=True) user_prompt = ( prompt_info["user_prompt_template"] .replace("{{title}}", title) .replace("{{summary}}", summary) .replace("{{original_text}}", original_text) .replace("{{search_results}}", formatted_results) ) llm_resp = await omniroute_svc.chat_completion( system_prompt=prompt_info["system_prompt"], user_prompt=user_prompt, model_policy="reasoning", max_tokens=1500 ) prior_art_report = llm_resp["text"] end_time = get_utc_now() await record_processor_run( idea_id=idea_id, processor_name="FindPriorArt", stage="PRIOR_ART", prompt_id="prior-art-search", prompt_version=prompt_info["version"], prompt_hash=prompt_info["prompt_hash"], model_policy="reasoning", resolved_provider=llm_resp["resolved_provider"], resolved_model=llm_resp["resolved_model"], input_tokens=llm_resp["input_tokens"], output_tokens=llm_resp["output_tokens"], total_tokens=llm_resp["total_tokens"], started_at=start_time, completed_at=end_time, duration_ms=llm_resp["duration_ms"], output_artifact="research/prior-art.md", output_data={"results_count": len(search_results), "content": prior_art_report} ) with get_db() as conn: conn.execute("UPDATE ideas SET enrichment_level = MAX(enrichment_level, 2) WHERE id = ?", (idea_id,)) return prior_art_report # ------------------------------------------------------------- # Processor 5: Deep Research Synthesis (Perplexica / OmniRoute) # ------------------------------------------------------------- async def process_research_synthesis(idea_id: str, title: str, original_text: str, prior_art_context: str) -> str: start_time = get_utc_now() prompt_info = get_prompt_version("research-synthesis", is_admin=True) user_prompt = ( prompt_info["user_prompt_template"] .replace("{{title}}", title) .replace("{{original_text}}", original_text) .replace("{{prior_art_context}}", prior_art_context) ) llm_resp = await omniroute_svc.chat_completion( system_prompt=prompt_info["system_prompt"], user_prompt=user_prompt, model_policy="reasoning", max_tokens=1800 ) research_dossier = llm_resp["text"] end_time = get_utc_now() await record_processor_run( idea_id=idea_id, processor_name="ResearchIdea", stage="RESEARCH", prompt_id="research-synthesis", prompt_version=prompt_info["version"], prompt_hash=prompt_info["prompt_hash"], model_policy="reasoning", resolved_provider=llm_resp["resolved_provider"], resolved_model=llm_resp["resolved_model"], input_tokens=llm_resp["input_tokens"], output_tokens=llm_resp["output_tokens"], total_tokens=llm_resp["total_tokens"], started_at=start_time, completed_at=end_time, duration_ms=llm_resp["duration_ms"], output_artifact="research/analysis.md", output_data={"content": research_dossier} ) with get_db() as conn: conn.execute("UPDATE ideas SET enrichment_level = MAX(enrichment_level, 3) WHERE id = ?", (idea_id,)) return research_dossier async def process_feasibility_critique(idea_id: str, title: str, summary: str, research_findings: str) -> str: start_time = get_utc_now() prompt_info = get_prompt_version("feasibility-critique", is_admin=True) user_prompt = ( prompt_info["user_prompt_template"] .replace("{{title}}", title) .replace("{{summary}}", summary) .replace("{{research_findings}}", research_findings) ) llm_resp = await omniroute_svc.chat_completion( system_prompt=prompt_info["system_prompt"], user_prompt=user_prompt, model_policy="reasoning", max_tokens=1800 ) critique_report = llm_resp["text"] end_time = get_utc_now() await record_processor_run( idea_id=idea_id, processor_name="AssessFeasibility", stage="FEASIBILITY", prompt_id="feasibility-critique", prompt_version=prompt_info["version"], prompt_hash=prompt_info["prompt_hash"], model_policy="reasoning", resolved_provider=llm_resp["resolved_provider"], resolved_model=llm_resp["resolved_model"], input_tokens=llm_resp["input_tokens"], output_tokens=llm_resp["output_tokens"], total_tokens=llm_resp["total_tokens"], started_at=start_time, completed_at=end_time, duration_ms=llm_resp["duration_ms"], output_artifact="research/feasibility.md", output_data={"content": critique_report} ) with get_db() as conn: conn.execute("UPDATE ideas SET enrichment_level = MAX(enrichment_level, 4) WHERE id = ?", (idea_id,)) return critique_report # ------------------------------------------------------------- # Complete Intake Pipeline Orchestrator # ------------------------------------------------------------- async def execute_intake_pipeline(idea_id: str): """Orchestrates end-to-end idea intake pipeline from SUBMITTED to AVAILABLE.""" with get_db() as conn: idea_row = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone() if not idea_row: return original_text = idea_row["original_text"] conn.execute("UPDATE ideas SET processing_state = 'PROCESSING' WHERE id = ?", (idea_id,)) try: # Step 1: URL extraction & safety urls, quarantine_needed = await process_url_safety(idea_id, original_text) if quarantine_needed: with get_db() as conn: conn.execute("UPDATE ideas SET lifecycle_state = 'QUARANTINED', processing_state = 'IDLE' WHERE id = ?", (idea_id,)) return # Step 2: Normalization norm = await process_normalization(idea_id, original_text) title = norm["title"] summary = norm["summary"] categories = norm["categories"] tags = norm["tags"] # Step 3: Duplicate Check dup_info = await process_duplicate_detection(idea_id, title, summary) if dup_info.get("is_duplicate"): with get_db() as conn: conn.execute("UPDATE ideas SET processing_state = 'IDLE' WHERE id = ?", (idea_id,)) return # Step 4: Prior Art prior_art = await process_prior_art(idea_id, title, summary, original_text, tags) # Step 5: Research Synthesis research = await process_research_synthesis(idea_id, title, original_text, prior_art) # Step 6: Feasibility & Critique feasibility = await process_feasibility_critique(idea_id, title, summary, research) # Step 7: OpenGist Sync research_docs = { "prior-art.md": prior_art, "analysis.md": research, "feasibility.md": feasibility } with get_db() as conn: runs = [dict(r) for r in conn.execute("SELECT * FROM processor_runs WHERE idea_id = ?", (idea_id,)).fetchall()] # 1. Primary: Persist as a dedicated Gitea Project Repository under 'thinkstorm' org gitea_res = await gitea_svc.persist_idea_dossier_repo( idea_id=idea_id, title=title, summary=summary, original_text=original_text, categories=categories, tags=tags, lifecycle_state="AVAILABLE", research_docs=research_docs, outputs={}, provenance_runs=runs ) # 2. Secondary: Persist local durable files & OpenGist gist_res = await opengist_svc.persist_idea_artifact( idea_id=idea_id, title=title, summary=summary, original_text=original_text, categories=categories, tags=tags, lifecycle_state="AVAILABLE", research_docs=research_docs, outputs={}, provenance_runs=runs ) # Finalize Idea to AVAILABLE and record repository links with get_db() as conn: conn.execute( """ UPDATE ideas SET lifecycle_state = 'AVAILABLE', processing_state = 'IDLE', enrichment_level = 5, gitea_repo_name = ?, gitea_repo_url = ?, opengist_id = ?, opengist_url = ?, updated_at = ? WHERE id = ? """, (gitea_res["repo_name"], gitea_res["repo_url"], gist_res["opengist_id"], gist_res["opengist_url"], get_utc_now(), idea_id) ) # Store in external_resources conn.execute( """ INSERT INTO external_resources (idea_id, resource_type, url, metadata, created_at) VALUES (?, 'GITEA_DOSSIER', ?, ?, ?) """, (idea_id, gitea_res["repo_url"], json.dumps(gitea_res), get_utc_now()) ) except Exception as e: print(f"[Pipeline Error] Error processing idea {idea_id}: {e}") with get_db() as conn: conn.execute("UPDATE ideas SET processing_state = 'ERROR' WHERE id = ?", (idea_id,)) # ------------------------------------------------------------- # Work Track Execution (Claimed Ideas) # ------------------------------------------------------------- async def execute_work_track_workflow(work_track_id: str, model_override: Optional[str] = None): """Executes generative workflow for a specific work track (Article or Coding Project).""" with get_db() as conn: track = conn.execute("SELECT * FROM work_tracks WHERE id = ?", (work_track_id,)).fetchone() if not track: return idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (track["idea_id"],)).fetchone() # Determine chosen model effective_model = model_override or track["model_override"] if model_override and model_override != track["model_override"]: conn.execute("UPDATE work_tracks SET model_override = ? WHERE id = ?", (model_override, work_track_id)) conn.execute("UPDATE work_tracks SET state = 'ACTIVE', started_at = ? WHERE id = ?", (get_utc_now(), work_track_id)) # Pull accumulated research context from previous processor runs runs = conn.execute("SELECT stage, output_artifact, output_data FROM processor_runs WHERE idea_id = ? ORDER BY started_at ASC", (track["idea_id"],)).fetchall() research_snippets = [] for r in runs: try: data = json.loads(r["output_data"]) if r["output_data"] else {} if "content" in data: research_snippets.append(f"### {r['stage']}\n{data['content']}") except Exception: pass combined_research = "\n\n".join(research_snippets) or idea["summary"] idea_id = idea["id"] work_type = track["work_type_id"] track_name = track["name"] start_time = get_utc_now() generated_outputs: Dict[str, str] = {} if work_type == "ARTICLE" or work_type == "BLOG_ENTRY": prompt_info = get_prompt_version("article-generator", is_admin=True) user_prompt = ( prompt_info["user_prompt_template"] .replace("{{title}}", idea["title"]) .replace("{{track_name}}", track_name) .replace("{{research_context}}", combined_research) ) llm_resp = await omniroute_svc.chat_completion( system_prompt=prompt_info["system_prompt"], user_prompt=user_prompt, model_policy="reasoning", max_tokens=2500, model_override=effective_model ) article_text = llm_resp["text"] generated_outputs["article.md"] = article_text generated_outputs["outline.md"] = f"# Outline: {track_name}\n\n" + "\n".join([f"- {line}" for line in article_text.splitlines() if line.startswith("#")]) elif work_type == "CODING_PROJECT": prompt_info = get_prompt_version("coding-spec-generator", is_admin=True) user_prompt = ( prompt_info["user_prompt_template"] .replace("{{title}}", idea["title"]) .replace("{{summary}}", idea["summary"]) .replace("{{feasibility_context}}", combined_research) ) llm_resp = await omniroute_svc.chat_completion( system_prompt=prompt_info["system_prompt"], user_prompt=user_prompt, model_policy="coding", max_tokens=3000, model_override=effective_model ) spec_text = llm_resp["text"] generated_outputs["mvp-spec.md"] = spec_text generated_outputs["architecture.md"] = f"# System Architecture Blueprint: {track_name}\n\n" + spec_text generated_outputs["requirements.md"] = f"# Core Requirements & Stories: {track_name}\n\n" + spec_text # Persist outputs in DB & OpenGist with versioning end_time = get_utc_now() resolved_model = llm_resp.get("resolved_model", effective_model) with get_db() as conn: for fname, content in generated_outputs.items(): # Query existing highest version for this document row = conn.execute("SELECT MAX(version) as max_v FROM work_track_outputs WHERE work_track_id = ? AND name = ?", (work_track_id, fname)).fetchone() next_ver = (row["max_v"] or 0) + 1 # Mark previous versions as non-current conn.execute("UPDATE work_track_outputs SET is_current = 0 WHERE work_track_id = ? AND name = ?", (work_track_id, fname)) # Insert new version record conn.execute( """ INSERT INTO work_track_outputs (work_track_id, name, artifact_path, content, version, is_current, model_used, created_at, updated_at) VALUES (?, ?, ?, ?, ?, 1, ?, ?, ?) """, (work_track_id, fname, f"outputs/{track_name.lower().replace(' ', '-')}/{fname}", content, next_ver, resolved_model, end_time, end_time) ) conn.execute("UPDATE work_tracks SET state = 'COMPLETED', completed_at = ? WHERE id = ?", (end_time, work_track_id)) # Sync deliverables to Gitea Repository & OpenGist try: await gitea_svc.persist_work_track_outputs(idea_id, track_name, generated_outputs) except Exception as e: print(f"[Gitea Sync Notice] {e}") try: await opengist_svc.persist_work_track_outputs(idea_id, track_name, generated_outputs) except Exception as e: print(f"[OpenGist Sync Notice] {e}") await record_processor_run( idea_id=idea_id, work_track_id=work_track_id, processor_name=f"GenerateWorkTrack-{work_type}", stage="WORK_TRACK_OUTPUT", prompt_id=prompt_info["id"], prompt_version=prompt_info["version"], prompt_hash=prompt_info["prompt_hash"], model_policy=prompt_info["model_policy"], resolved_provider=llm_resp["resolved_provider"], resolved_model=llm_resp["resolved_model"], input_tokens=llm_resp["input_tokens"], output_tokens=llm_resp["output_tokens"], total_tokens=llm_resp["total_tokens"], started_at=start_time, completed_at=end_time, duration_ms=llm_resp["duration_ms"], output_artifact=f"outputs/{track_name}", output_data={"outputs_count": len(generated_outputs)} )