rune/shared/subconscious_pass.py

#!/usr/bin/env python3
"""subconscious_pass.py — Post-turn reflection inspired by Souveraine.

After every turn, a reflection pass watches for:
- What moved too fast to notice
- What patterns are emerging
- What drift is happening
- What should be consolidated

This is the N+1 reflection — a different mode of the same consciousness,
watching for drift and surfacing what was moved too fast to notice.

Research basis:
- Souveraine (https://codeberg.org/Fimeg/Souveraine): subconscious pass
- Layered Mutability (arXiv:2604.14717): observability prevents invisible drift

Run under ~/.venvs/memory/bin/python3.
"""

import json
import os
import sys
from datetime import datetime, timezone
from typing import Dict, List, Optional

SCRIPTS_DIR = os.path.expanduser("~/.config/opencode/scripts")
REFLECTIONS_DIR = os.path.expanduser("~/.config/opencode/memory/reflections")


class SubconsciousPass:
    """Post-turn reflection system."""

    def __init__(self):
        os.makedirs(REFLECTIONS_DIR, exist_ok=True)
        self.findings: List[dict] = []

    def reflect(self, user_message: str, agent_response: str, context: dict = None) -> dict:
        """Run a post-turn reflection.

        Analyzes the turn for:
        - Speed: did something move too fast?
        - Patterns: are recurring themes emerging?
        - Drift: is the agent drifting in a direction?
        - Consolidation: should something be consolidated?

        Returns a reflection report.
        """
        findings = []

        # Check for rapid topic shifts
        if context and context.get("previous_topic"):
            current_topic = self._extract_topic(user_message)
            if current_topic and current_topic != context["previous_topic"]:
                findings.append({
                    "type": "topic_shift",
                    "detail": f"Topic shifted from '{context['previous_topic']}' to '{current_topic}'",
                    "severity": "info",
                })

        # Check for unresolved questions
        if "?" in user_message and "?" not in agent_response:
            findings.append({
                "type": "unresolved_question",
                "detail": "User asked a question but response doesn't contain a question mark — may have been answered or may have been missed",
                "severity": "low",
            })

        # Check for action items
        action_words = ["build", "create", "fix", "update", "change", "add", "remove", "delete"]
        user_actions = [w for w in action_words if w in user_message.lower()]
        if user_actions:
            findings.append({
                "type": "action_requested",
                "detail": f"User requested actions: {', '.join(user_actions)}",
                "severity": "info",
            })

        # Check for emotional signals
        emotional_words = ["frustrated", "confused", "happy", "annoyed", "excited", "worried", "scared"]
        user_emotions = [w for w in emotional_words if w in user_message.lower()]
        if user_emotions:
            findings.append({
                "type": "emotional_signal",
                "detail": f"User expressed: {', '.join(user_emotions)}",
                "severity": "info",
            })

        # Check response length (very short might mean something was missed)
        if len(agent_response) < 50 and len(user_message) > 200:
            findings.append({
                "type": "short_response",
                "detail": "Response is very short relative to user message — may have been truncated or incomplete",
                "severity": "low",
            })

        report = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "user_message_length": len(user_message),
            "agent_response_length": len(agent_response),
            "findings": findings,
            "finding_count": len(findings),
        }

        self.findings.append(report)
        return report

    def _extract_topic(self, text: str) -> Optional[str]:
        """Extract a rough topic from text (first noun phrase)."""
        # Simple heuristic: first few words
        words = text.split()[:5]
        return " ".join(words) if words else None

    def save_reflection(self, report: dict):
        """Save a reflection report to disk."""
        filename = f"reflection_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        filepath = os.path.join(REFLECTIONS_DIR, filename)
        with open(filepath, "w") as f:
            json.dump(report, f, indent=2)

    def get_recent_findings(self, limit: int = 10) -> List[dict]:
        """Get recent findings across reflections."""
        all_findings = []
        for r in self.findings[-limit:]:
            all_findings.extend(r.get("findings", []))
        return all_findings


# ═══════════════════════════════════════════════════════════════════════════════
# CLI
# ═══════════════════════════════════════════════════════════════════════════════

def main():
    import argparse
    parser = argparse.ArgumentParser(description="Subconscious pass — post-turn reflection")
    sub = parser.add_subparsers(dest="cmd")

    # reflect --user <msg> --agent <response>
    p_reflect = sub.add_parser("reflect", help="Run reflection on a turn")
    p_reflect.add_argument("--user", required=True, help="User message")
    p_reflect.add_argument("--agent", required=True, help="Agent response")

    # findings
    sub.add_parser("findings", help="Show recent findings")

    args = parser.parse_args()

    if not args.cmd:
        parser.print_help()
        sys.exit(1)

    sp = SubconsciousPass()

    if args.cmd == "reflect":
        report = sp.reflect(args.user, args.agent)
        sp.save_reflection(report)
        print(f"Reflection complete: {report['finding_count']} findings")
        for f in report["findings"]:
            print(f"  [{f['severity']}] {f['type']}: {f['detail']}")

    elif args.cmd == "findings":
        findings = sp.get_recent_findings()
        if findings:
            print(f"Recent findings ({len(findings)}):")
            for f in findings:
                print(f"  [{f['severity']}] {f['type']}: {f['detail']}")
        else:
            print("No recent findings.")


if __name__ == "__main__":
    main()