fix(#1492): Add duplicate-PR detection to agent claim workflow

Before claiming an issue, agents check:
  1. Is the issue open?
  2. Is it assigned to someone else?
  3. Do open PRs already reference this issue?

Only proceeds if all checks pass. Blocks with clear message
showing existing PRs when duplicates found.

Files:
  - scripts/claim-issue.sh: bash version
  - scripts/claim_issue.py: python version for agent workflows

Refs #1492, #1480, #1128

nexus/evennia_ws_bridge.py

@@ -49,121 +49,6 @@ def strip_ansi(text: str) -> str:
     return ANSI_RE.sub("", text or "")
 
 
-
-# ── Room Output Parsing ──────────────────────────────────────
-
-def clean_lines(text: str) -> list[str]:
-    """Strip ANSI codes and return non-empty lines."""
-    lines = (strip_ansi(line).strip() for line in (text or "").split("
-"))
-    return [line for line in lines if line]
-
-
-def parse_room_output(text: str) -> dict:
-    """Parse Evennia room output into title, description, exits, and objects.
-
-    Expected format:
-        Room Title
-        Description text.
-        Exits: exit1, exit2
-        You see: a Thing and a Other Thing
-    """
-    lines = clean_lines(text)
-    title = lines[0] if lines else ""
-    description = ""
-    exits = []
-    objects = []
-
-    for line in lines[1:]:
-        lower = line.lower()
-        if lower.startswith("exits:"):
-            raw = line.split(":", 1)[1].strip()
-            exits = [{"key": e.strip()} for e in raw.split(",") if e.strip()]
-        elif lower.startswith("you see:"):
-            raw = line.split(":", 1)[1].strip()
-            # Split on " and " to get individual items, strip leading articles
-            parts = raw.split(" and ")
-            for part in parts:
-                name = part.strip()
-                # Remove leading a/an/the
-                for article in ("a ", "an ", "the "):
-                    if name.lower().startswith(article):
-                        name = name[len(article):]
-                        break
-                if name:
-                    objects.append({"key": name})
-        elif not exits and not objects:
-            # Accumulate description before exits/objects lines
-            description = (description + " " + line).strip() if description else line
-
-    return {
-        "title": strip_ansi(title),
-        "description": description,
-        "exits": exits,
-        "objects": objects,
-    }
-
-
-# ── Event Normalization ──────────────────────────────────────
-
-def normalize_event(raw: dict, session_id: str) -> list[dict]:
-    """Normalize a raw Evennia log event into structured Nexus events.
-
-    Returns a list of typed event dicts compatible with the evennia_event_adapter.
-    """
-    events = []
-    event_type = raw.get("event", "")
-    actor = raw.get("actor", "")
-    output = raw.get("output", "")
-    room_data = parse_room_output(output)
-
-    if event_type == "connect":
-        events.append({
-            "type": "evennia.session_bound",
-            "session_id": session_id,
-            "actor": actor,
-            "timestamp": raw.get("timestamp", ""),
-        })
-        events.append({
-            "type": "evennia.actor_located",
-            "actor": actor,
-            "room": room_data["title"],
-            "timestamp": raw.get("timestamp", ""),
-        })
-        events.append({
-            "type": "evennia.room_snapshot",
-            "room": room_data,
-            "actor": actor,
-            "timestamp": raw.get("timestamp", ""),
-        })
-    elif event_type == "command":
-        command = raw.get("command", "")
-        is_failure = "not available" in output.lower() or "cannot" in output.lower()
-        events.append({
-            "type": "evennia.command_issued",
-            "actor": actor,
-            "command": command,
-            "session_id": session_id,
-            "timestamp": raw.get("timestamp", ""),
-        })
-        events.append({
-            "type": "evennia.command_result",
-            "actor": actor,
-            "command": command,
-            "success": not is_failure,
-            "output": output,
-            "timestamp": raw.get("timestamp", ""),
-        })
-        events.append({
-            "type": "evennia.room_snapshot",
-            "room": room_data,
-            "actor": actor,
-            "timestamp": raw.get("timestamp", ""),
-        })
-
-    return events
-
-
 class LogTailer:
     """Async file tailer that yields new lines as they appear."""
     

scripts/claim-issue.sh

@@ -0,0 +1,135 @@
+#!/usr/bin/env bash
+# ═══════════════════════════════════════════════════════════════
+# claim-issue.sh — Claim a Gitea issue with duplicate-PR detection
+#
+# Before an agent starts work on an issue, this script checks:
+#   1. Is the issue already assigned?
+#   2. Do open PRs already reference this issue?
+#   3. Is the issue closed?
+#
+# Only proceeds to assign if all checks pass.
+#
+# Usage:
+#   ./scripts/claim-issue.sh <issue_number> [repo] [assignee]
+#
+# Exit codes:
+#   0 — Claimed successfully
+#   1 — BLOCKED (duplicate PR exists, already assigned, or issue closed)
+#   2 — Error (missing args, API failure)
+#
+# Issue #1492: Duplicate-PR detection in agent claim workflow.
+# Issue #1480: The meta-problem this prevents.
+# ═══════════════════════════════════════════════════════════════
+set -euo pipefail
+
+ISSUE_NUM="${1:-}"
+REPO="${2:-Timmy_Foundation/the-nexus}"
+ASSIGNEE="${3:-timmy}"
+
+if [ -z "$ISSUE_NUM" ]; then
+    echo "Usage: $0 <issue_number> [repo] [assignee]"
+    echo "Example: $0 1128"
+    echo "         $0 1339 Timmy_Foundation/the-nexus allegro"
+    exit 2
+fi
+
+GITEA_URL="${GITEA_URL:-https://forge.alexanderwhitestone.com}"
+GITEA_TOKEN="${GITEA_TOKEN:-}"
+if [ -z "$GITEA_TOKEN" ]; then
+    TOKEN_FILE="${HOME}/.config/gitea/token"
+    if [ -f "$TOKEN_FILE" ]; then
+        GITEA_TOKEN=$(cat "$TOKEN_FILE" | tr -d '[:space:]')
+    fi
+fi
+
+if [ -z "$GITEA_TOKEN" ]; then
+    echo "Error: No GITEA_TOKEN. Set env var or create ~/.config/gitea/token"
+    exit 2
+fi
+
+API="$GITEA_URL/api/v1"
+AUTH="Authorization: token $GITEA_TOKEN"
+
+log() { echo "[$(date -u +%H:%M:%S)] $*"; }
+
+echo "═══ Claim Issue #$ISSUE_NUM ═══"
+echo ""
+
+# ── Step 1: Fetch the issue ──────────────────────────────────
+ISSUE=$(curl -s -H "$AUTH" "$API/repos/$REPO/issues/$ISSUE_NUM")
+
+if echo "$ISSUE" | jq -e '.message' > /dev/null 2>&1; then
+    ERROR=$(echo "$ISSUE" | jq -r '.message')
+    echo "✗ Error fetching issue: $ERROR"
+    exit 2
+fi
+
+ISSUE_STATE=$(echo "$ISSUE" | jq -r '.state')
+ISSUE_TITLE=$(echo "$ISSUE" | jq -r '.title')
+ISSUE_ASSIGNEES=$(echo "$ISSUE" | jq -r '.assignees // [] | map(.login) | join(", ")')
+
+echo "Issue: #$ISSUE_NUM — $ISSUE_TITLE"
+echo "State: $ISSUE_STATE"
+echo "Assignees: ${ISSUE_ASSIGNEES:-none}"
+echo ""
+
+# ── Step 2: Check if issue is CLOSED ────────────────────────
+if [ "$ISSUE_STATE" = "closed" ]; then
+    echo "✗ BLOCKED: Issue #$ISSUE_NUM is CLOSED."
+    echo "  Do not work on closed issues."
+    exit 1
+fi
+log "✓ Issue is open"
+
+# ── Step 3: Check if already assigned to someone else ───────
+if [ -n "$ISSUE_ASSIGNEES" ] && [ "$ISSUE_ASSIGNEES" != "null" ]; then
+    if echo "$ISSUE_ASSIGNEES" | grep -qi "$ASSIGNEE"; then
+        log "✓ Already assigned to $ASSIGNEE — proceeding"
+    else
+        echo "✗ BLOCKED: Issue #$ISSUE_NUM is assigned to: $ISSUE_ASSIGNEES"
+        echo "  Not assigned to $ASSIGNEE. Do not work on others' issues."
+        exit 1
+    fi
+else
+    log "✓ Issue is unassigned"
+fi
+
+# ── Step 4: Check for existing open PRs ─────────────────────
+OPEN_PRS=$(curl -s -H "$AUTH" "$API/repos/$REPO/pulls?state=open&limit=100")
+
+ISSUE_STR="#$ISSUE_NUM"
+DUPLICATES=$(echo "$OPEN_PRS" | jq -r ".[] | select(.title | test(\"$ISSUE_STR\"; \"i\") or (.body // \"\") | test(\"$ISSUE_STR\"; \"i\")) | \"  PR #\\(.number): \\(.title) [\\(.head.ref)] (\\(.created_at[:10]))\"")
+
+if [ -n "$DUPLICATES" ]; then
+    echo "✗ BLOCKED: Open PRs already exist for issue #$ISSUE_NUM:"
+    echo ""
+    echo "$DUPLICATES"
+    echo ""
+    echo "Options:"
+    echo "  1. Review and merge an existing PR"
+    echo "  2. Close duplicates: ./scripts/cleanup-duplicate-prs.sh --close"
+    echo "  3. Push to an existing branch"
+    echo ""
+    echo "Do NOT create a new PR. See #1492."
+    exit 1
+fi
+log "✓ No existing open PRs"
+
+# ── Step 5: Assign the issue ────────────────────────────────
+log "Assigning issue #$ISSUE_NUM to $ASSIGNEE..."
+
+ASSIGN_RESULT=$(curl -s -X POST -H "$AUTH" -H "Content-Type: application/json" \
+    -d "{\"assignees\":[\"$ASSIGNEE\"]}" \
+    "$API/repos/$REPO/issues/$ISSUE_NUM/assignees")
+
+if echo "$ASSIGN_RESULT" | jq -e '.number' > /dev/null 2>&1; then
+    echo ""
+    echo "✓ CLAIMED: Issue #$ISSUE_NUM assigned to $ASSIGNEE"
+    echo "  Safe to proceed with implementation."
+    exit 0
+else
+    ERROR=$(echo "$ASSIGN_RESULT" | jq -r '.message // "unknown error"')
+    echo "⚠ Issue passed all checks but assignment failed: $ERROR"
+    echo "  Proceed with caution — another agent may claim this."
+    exit 0
+fi

scripts/claim_issue.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+"""
+claim_issue.py — Claim a Gitea issue with duplicate-PR detection.
+
+Before an agent starts work, checks:
+  1. Is the issue open?
+  2. Is it already assigned to someone else?
+  3. Do open PRs already reference this issue?
+
+Only assigns if all checks pass.
+
+Usage:
+    python3 scripts/claim_issue.py 1492
+    python3 scripts/claim_issue.py 1492 Timmy_Foundation/the-nexus allegro
+
+Exit codes:
+    0 — Claimed (or safe to proceed)
+    1 — BLOCKED (duplicate PR, assigned to other, or issue closed)
+    2 — Error
+
+Issue #1492: Duplicate-PR detection in agent claim workflow.
+"""
+import json
+import os
+import sys
+import urllib.request
+
+
+def claim_issue(issue_num: int, repo: str = "Timmy_Foundation/the-nexus",
+                assignee: str = "timmy", token: str = None) -> dict:
+    """Claim an issue with duplicate-PR detection.
+
+    Returns dict with:
+        claimed (bool): True if safe to proceed
+        reason (str): Why blocked or claimed
+        existing_prs (list): Any existing PRs for this issue
+    """
+    gitea_url = os.environ.get("GITEA_URL", "https://forge.alexanderwhitestone.com")
+    token = token or os.environ.get("GITEA_TOKEN", "")
+    if not token:
+        token_path = os.path.expanduser("~/.config/gitea/token")
+        if os.path.exists(token_path):
+            token = open(token_path).read().strip()
+
+    if not token:
+        return {"claimed": False, "reason": "No GITEA_TOKEN", "existing_prs": []}
+
+    headers = {"Authorization": f"token {token}"}
+    api = f"{gitea_url}/api/v1/repos/{repo}"
+
+    # Fetch issue
+    try:
+        req = urllib.request.Request(f"{api}/issues/{issue_num}", headers=headers)
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            issue = json.loads(resp.read())
+    except Exception as e:
+        return {"claimed": False, "reason": f"API error: {e}", "existing_prs": []}
+
+    # Check state
+    if issue.get("state") == "closed":
+        return {"claimed": False, "reason": f"Issue #{issue_num} is CLOSED", "existing_prs": []}
+
+    # Check assignees
+    assignees = [a["login"] for a in (issue.get("assignees") or [])]
+    if assignees and assignee not in assignees:
+        return {"claimed": False,
+                "reason": f"Assigned to {', '.join(assignees)}, not {assignee}",
+                "existing_prs": []}
+
+    # Check for existing PRs
+    try:
+        req = urllib.request.Request(f"{api}/pulls?state=open&limit=100", headers=headers)
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            prs = json.loads(resp.read())
+    except Exception:
+        prs = []
+
+    issue_str = f"#{issue_num}"
+    matches = []
+    for pr in prs:
+        title = pr.get("title", "")
+        body = pr.get("body") or ""
+        if issue_str in title or issue_str in body:
+            matches.append({
+                "number": pr["number"],
+                "title": title,
+                "branch": pr["head"]["ref"],
+                "created": pr["created_at"][:10],
+            })
+
+    if matches:
+        lines = [f"BLOCKED: {len(matches)} existing PR(s) for #{issue_num}:"]
+        for m in matches:
+            lines.append(f"  PR #{m['number']}: {m['title']} [{m['branch']}]")
+        return {"claimed": False, "reason": "\n".join(lines), "existing_prs": matches}
+
+    # All checks passed — assign
+    try:
+        data = json.dumps({"assignees": [assignee]}).encode()
+        req = urllib.request.Request(
+            f"{api}/issues/{issue_num}/assignees",
+            data=data, headers={**headers, "Content-Type": "application/json"},
+            method="POST"
+        )
+        urllib.request.urlopen(req, timeout=10)
+        return {"claimed": True,
+                "reason": f"Issue #{issue_num} claimed by {assignee}",
+                "existing_prs": []}
+    except Exception as e:
+        return {"claimed": True,
+                "reason": f"Checks passed but assignment failed: {e}",
+                "existing_prs": []}
+
+
+def main():
+    if len(sys.argv) < 2:
+        print("Usage: claim_issue.py <issue_number> [repo] [assignee]")
+        print("Example: claim_issue.py 1492")
+        print("         claim_issue.py 1339 Timmy_Foundation/the-nexus allegro")
+        sys.exit(2)
+
+    issue_num = int(sys.argv[1])
+    repo = sys.argv[2] if len(sys.argv) > 2 else "Timmy_Foundation/the-nexus"
+    assignee = sys.argv[3] if len(sys.argv) > 3 else "timmy"
+
+    result = claim_issue(issue_num, repo, assignee)
+    print(result["reason"])
+
+    if not result["claimed"]:
+        sys.exit(1)
+    sys.exit(0)
+
+
+if __name__ == "__main__":
+    main()