Grammar Transformation Guide

def get_tasks():
    """Get today's tasks."""
    tasks = db.query(Task).filter(Task.due_date == date.today()).all()
    return {"tasks": [t.to_dict() for t in tasks]}

async def perceive_todays_moments(user: Entity, workspace: Place) -> Perception:
    """
    User experiences today's task Moments through Temporal lens.

    Applies Pattern-050 (Context/Result pair) and Pattern-052 (Personality Bridge).
    """
    # Context: What we're observing
    with Situation(
        entities=[user, piper],
        tension="What needs attention today?",
        place=workspace
    ) as situation:
        # Gather Moments from Place with lens
        tasks = await gather_from_place(
            place=workspace,
            lens=TemporalLens(mode=PerceptionMode.NOTICING),
            filter_context={"timeframe": "today"}
        )

        # Personality Bridge: Transform data into awareness
        framing = _frame_tasks_with_warmth(tasks, situation)

        # Return Perception (not raw data)
        return Perception(
            framing=framing,  # "I notice you have 3 things that want attention today"
            moments=tasks,
            situation=situation,
            lens_applied=TemporalLens.__name__
        )

def _frame_tasks_with_warmth(tasks: List[Task], situation: Situation) -> str:
    """
    Apply Pattern-052 (Personality Bridge) and Pattern-053 (Warmth Calibration).
    """
    count = len(tasks)

    if count == 0:
        return "You have a clear day ahead - no pressing tasks right now."
    elif count <= 3:
        return f"I notice {count} things that want your attention today."
    else:
        return f"You have {count} tasks for today. Would you like me to help prioritize?"

from services.mux.protocols import EntityProtocol, MomentProtocol, PlaceProtocol

# Entity has identity and can experience
class User(EntityProtocol):
    id: str
    name: str
    def experiences(self, moment: MomentProtocol) -> Perception:
        ...

# Moment has timestamp and captures significance
class Meeting(MomentProtocol):
    id: str
    timestamp: datetime
    def captures(self) -> Dict[str, Any]:
        return {"attendees": [...], "outcomes": [...]}

# Place has atmosphere and contains things
class GitHubRepo(PlaceProtocol):
    id: str
    atmosphere: str = "technical review space"
    def contains(self) -> List[Any]:
        return [prs, issues, commits]

from services.mux.lenses import (
    TemporalLens, PriorityLens, CollaborativeLens, FlowLens,
    HierarchyLens, QuantitativeLens, CausalLens, ContextualLens
)
from services.mux.protocols import PerceptionMode

# Time-aware query
temporal = TemporalLens(mode=PerceptionMode.NOTICING)
tasks_today = temporal.perceive(workspace, filter={"timeframe": "today"})

# Priority-aware query
priority = PriorityLens(mode=PerceptionMode.ANTICIPATING)
urgent_items = priority.perceive(workspace, threshold="high")

# People-aware query
collab = CollaborativeLens(mode=PerceptionMode.NOTICING)
team_activity = collab.perceive(workspace, scope="team")

# Progress-aware query
flow = FlowLens(mode=PerceptionMode.NOTICING)
blocked_items = flow.perceive(workspace, state="blocked")

async def _handle_stale_prs(self, intent: Intent, workflow_id: str) -> IntentProcessingResult:
    """
    Handle "Show me stale PRs" query.
    Returns open PRs older than 7 days with age and title.
    """
    self.logger.info(f"Processing stale PRs query: {intent.action}")

    try:
        # Import and initialize GitHub router
        from services.integrations.github.github_integration_router import GitHubIntegrationRouter
        github_router = GitHubIntegrationRouter()
        await github_router.initialize()

        # Check if GitHub is configured
        if not github_router.config_service.is_configured():
            return IntentProcessingResult(
                success=True,
                message=(
                    "I'd love to show you stale PRs, but GitHub isn't configured yet. "
                    "To enable GitHub integration, please add your GITHUB_TOKEN to your environment "
                    "or configure it in PIPER.user.md. Once configured, I can track open PRs!"
                ),
                intent_data={
                    "category": intent.category.value,
                    "action": intent.action,
                    "confidence": intent.confidence,
                },
                workflow_id=workflow_id,
                requires_clarification=False,
                implemented=False,
            )

        # Get open issues (includes PRs)
        from datetime import datetime, timedelta, timezone
        open_items = await github_router.get_open_issues(limit=100)

        # Filter to PRs older than 7 days
        now = datetime.now(timezone.utc)
        stale_threshold = now - timedelta(days=7)

        stale_prs = []
        for item in open_items:
            if not item.get("pull_request"):
                continue

            created_str = item.get("created_at")
            created_dt = datetime.fromisoformat(created_str.replace("Z", "+00:00"))

            if created_dt < stale_threshold:
                age_days = (now - created_dt).days
                stale_prs.append({
                    "title": item.get("title"),
                    "number": item.get("number"),
                    "age_days": age_days,
                    "url": item.get("html_url"),
                    "author": item.get("user", {}).get("login", "Unknown"),
                })

        # Format response
        if not stale_prs:
            message = "Great news! You don't have any stale PRs right now. All your open PRs are recent."
        else:
            pr_list = "\n".join([
                f"- #{pr['number']}: {pr['title']} ({pr['age_days']} days old, by {pr['author']})"
                for pr in stale_prs
            ])
            message = f"I found {len(stale_prs)} stale PR(s) (older than 7 days):\n\n{pr_list}"

        return IntentProcessingResult(
            success=True,
            message=message,
            intent_data={
                "category": intent.category.value,
                "action": intent.action,
                "stale_prs": stale_prs,
                "count": len(stale_prs),
            },
            workflow_id=workflow_id,
        )

    except Exception as e:
        self.logger.error(f"Error processing stale PRs: {e}")
        return IntentProcessingResult(
            success=False,
            message="I ran into a problem checking for stale PRs. Please try again.",
            intent_data={"error": str(e)},
            workflow_id=workflow_id,
            error=str(e),
        )

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import List, Optional

from services.mux.protocols import EntityProtocol, MomentProtocol, PlaceProtocol, PerceptionMode
from services.mux.lenses import TemporalLens, CollaborativeLens, FlowLens
from services.mux.situation import Situation, Perception


# Pattern-050: Context Dataclass Pair
@dataclass
class StalePRsContext:
    """
    Input context for stale PR perception.
    Preserves Entity/Moment/Place throughout flow.
    """
    user: EntityProtocol  # Who's asking
    github_place: PlaceProtocol  # Where we're looking
    stale_threshold_days: int = 7  # What defines "stale"
    requested_at: datetime = None  # When they asked

    def __post_init__(self):
        if self.requested_at is None:
            self.requested_at = datetime.now(timezone.utc)


@dataclass
class StalePRsResult:
    """
    Output perception of stale PRs.
    Grammar-conscious result with framing.
    """
    success: bool
    framing: str  # Narrative opening (Pattern-052: Personality Bridge)
    stale_moments: List['PRMoment']  # Not "items" or "records"
    situation: Situation  # Context preserved
    lens_applied: List[str]  # Which perspectives used
    learning: Optional[str] = None  # Extracted wisdom


@dataclass
class PRMoment(MomentProtocol):
    """
    A PR as a Moment (bounded significant occurrence).
    Not just data - it's a scene in the collaboration.
    """
    id: str  # PR number
    timestamp: datetime  # When created
    title: str
    author: EntityProtocol  # Not just a string
    age_days: int
    url: str

    def captures(self) -> dict:
        """What this Moment captured (policy, process, people, outcomes)."""
        return {
            "policy": "Code review process",
            "process": "Pull request workflow",
            "people": [self.author],
            "outcomes": "Awaiting review or merge",
        }

    def narrative(self) -> str:
        """Pattern-052: Transform into narrative description."""
        if self.age_days <= 3:
            age_desc = "from a few days ago"
        elif self.age_days <= 7:
            age_desc = "from about a week ago"
        elif self.age_days <= 14:
            age_desc = "from a couple weeks ago"
        else:
            age_desc = f"from {self.age_days} days ago"

        return f"'{self.title}' by {self.author.name} {age_desc}"


async def perceive_stale_prs(context: StalePRsContext) -> StalePRsResult:
    """
    User experiences stale PR Moments in GitHub Place through multiple Lenses.

    Applies:
    - Pattern-050: Context/Result pair preserves grammar
    - Pattern-051: Parallel Place Gathering (if multiple repos)
    - Pattern-052: Personality Bridge for framing
    - Pattern-053: Warmth Calibration based on count
    - Pattern-054: Honest Failure if GitHub unavailable
    """
    # Frame the Situation
    with Situation(
        entities=[context.user, piper],
        place=context.github_place,
        tension="Old work might need attention or closure",
        learning_question="What patterns emerge in PR staleness?"
    ) as situation:

        # Pattern-054: Honest Failure if Place not accessible
        if not await context.github_place.is_accessible():
            return _create_honest_failure_result(context, situation)

        # Gather Moments from Place using multiple Lenses
        stale_moments = await _gather_stale_pr_moments(
            context=context,
            situation=situation
        )

        # Pattern-052: Personality Bridge - transform data into awareness
        framing = _frame_stale_prs_with_warmth(
            moments=stale_moments,
            context=context,
            situation=situation
        )

        # Extract learning (for future composting)
        learning = _extract_staleness_patterns(stale_moments)

        return StalePRsResult(
            success=True,
            framing=framing,
            stale_moments=stale_moments,
            situation=situation,
            lens_applied=["Temporal", "Collaborative", "Flow"],
            learning=learning
        )


async def _gather_stale_pr_moments(
    context: StalePRsContext,
    situation: Situation
) -> List[PRMoment]:
    """
    Gather stale PR Moments from GitHub Place using multiple Lenses.
    """
    # Apply Temporal Lens: perceive PRs through time
    temporal_lens = TemporalLens(mode=PerceptionMode.REMEMBERING)
    all_open_prs = await temporal_lens.perceive(
        place=context.github_place,
        filter={"state": "open", "type": "pull_request"}
    )

    # Apply Flow Lens: understand status
    flow_lens = FlowLens(mode=PerceptionMode.NOTICING)

    # Apply Collaborative Lens: understand people
    collab_lens = CollaborativeLens(mode=PerceptionMode.NOTICING)

    # Filter to stale (older than threshold)
    now = datetime.now(timezone.utc)
    stale_threshold = now - timedelta(days=context.stale_threshold_days)

    stale_moments = []
    for pr_data in all_open_prs:
        created_dt = datetime.fromisoformat(pr_data["created_at"].replace("Z", "+00:00"))

        if created_dt < stale_threshold:
            # Transform data item into Moment (with Entity author)
            author_entity = _create_author_entity(pr_data.get("user", {}))
            age_days = (now - created_dt).days

            moment = PRMoment(
                id=str(pr_data["number"]),
                timestamp=created_dt,
                title=pr_data["title"],
                author=author_entity,
                age_days=age_days,
                url=pr_data["html_url"]
            )

            stale_moments.append(moment)

    return stale_moments


def _create_author_entity(user_data: dict) -> EntityProtocol:
    """
    Create Entity for PR author (not just a string).
    Preserves identity and potential for agency.
    """
    @dataclass
    class GitHubUser(EntityProtocol):
        id: str
        name: str
        profile_url: str

        def experiences(self, moment: MomentProtocol):
            # Authors experience PR waiting as tension
            return f"{self.name} is waiting for review"

    return GitHubUser(
        id=user_data.get("id", "unknown"),
        name=user_data.get("login", "Unknown"),
        profile_url=user_data.get("html_url", "")
    )


def _frame_stale_prs_with_warmth(
    moments: List[PRMoment],
    context: StalePRsContext,
    situation: Situation
) -> str:
    """
    Pattern-052: Personality Bridge
    Pattern-053: Warmth Calibration

    Transform count into experience language with appropriate warmth.
    """
    count = len(moments)

    if count == 0:
        # Warmth: Celebrate the good news
        return (
            "Great news! You don't have any PRs that have been waiting more than "
            f"{context.stale_threshold_days} days. Everything in GitHub is moving along."
        )

    elif count == 1:
        # Warmth: Gentle single-item attention
        moment = moments[0]
        return (
            f"I notice one PR that's been waiting a while: {moment.narrative()}. "
            "Might be time to check in or close it."
        )

    elif count <= 3:
        # Warmth: Moderate multi-item attention
        return (
            f"I found {count} PRs in GitHub that have been open for more than "
            f"{context.stale_threshold_days} days. They might need a nudge or closure:"
        )

    else:
        # Warmth: Higher urgency for many items
        return (
            f"There are {count} PRs in GitHub that have been waiting a while "
            f"(more than {context.stale_threshold_days} days). This might be a good time "
            "to review what's active and what can be closed:"
        )


def _extract_staleness_patterns(moments: List[PRMoment]) -> Optional[str]:
    """
    Extract learning from staleness patterns (for composting/memory).
    """
    if not moments:
        return None

    # Simple pattern: identify if one author has multiple stale PRs
    author_counts = {}
    for moment in moments:
        author_name = moment.author.name
        author_counts[author_name] = author_counts.get(author_name, 0) + 1

    multi_stale_authors = [name for name, count in author_counts.items() if count > 1]

    if multi_stale_authors:
        return (
            f"Pattern noticed: {', '.join(multi_stale_authors)} has multiple stale PRs. "
            "Might indicate bandwidth constraints or review bottleneck."
        )

    return "No significant patterns in staleness yet."


def _create_honest_failure_result(
    context: StalePRsContext,
    situation: Situation
) -> StalePRsResult:
    """
    Pattern-054: Honest Failure with Suggestion

    GitHub not accessible - provide honest acknowledgment with suggestion.
    """
    return StalePRsResult(
        success=False,
        framing=(
            "I'd love to check for stale PRs in GitHub, but I can't reach it right now. "
            "This might mean GitHub isn't configured yet, or there's a temporary connection issue. "
            "\n\n"
            "To enable GitHub integration, add your GITHUB_TOKEN to your environment "
            "or configure it in PIPER.user.md. Once configured, I can track PRs across time."
        ),
        stale_moments=[],
        situation=situation,
        lens_applied=["Temporal"],
        learning="GitHub Place not accessible during query"
    )


# Integration point: Route handler calls this
async def handle_stale_prs_intent(
    intent: Intent,
    user: EntityProtocol,
    github_place: PlaceProtocol,
    workflow_id: str
) -> IntentProcessingResult:
    """
    Route handler adapter (connects to existing IntentService).

    This is where the old flattened handler would call.
    Now it builds Context and calls grammar-conscious function.
    """
    # Build Context
    context = StalePRsContext(
        user=user,
        github_place=github_place,
        stale_threshold_days=7
    )

    # Perceive with grammar
    result = await perceive_stale_prs(context)

    # Transform to IntentProcessingResult (existing interface)
    return IntentProcessingResult(
        success=result.success,
        message=_format_message_for_user(result),
        intent_data={
            "category": intent.category.value,
            "action": intent.action,
            "stale_moments": [
                {
                    "title": m.title,
                    "author": m.author.name,
                    "age_days": m.age_days,
                    "narrative": m.narrative(),
                    "url": m.url
                }
                for m in result.stale_moments
            ],
            "count": len(result.stale_moments),
            "learning": result.learning,
            "lens_applied": result.lens_applied,
        },
        workflow_id=workflow_id,
        implemented=True
    )


def _format_message_for_user(result: StalePRsResult) -> str:
    """
    Format final message for user.
    Combines framing with moment narratives.
    """
    parts = [result.framing]

    if result.stale_moments:
        parts.append("")  # Blank line
        for moment in result.stale_moments:
            parts.append(f"  • {moment.narrative()} - [View PR]({moment.url})")

        if result.learning:
            parts.append("")
            parts.append(f"💡 {result.learning}")

    return "\n".join(parts)