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

deploy.sh

@@ -1,9 +1,7 @@
 #!/usr/bin/env bash
 # deploy.sh — spin up (or update) the Nexus staging environment
-# Usage: ./deploy.sh          — rebuild and restart nexus-main (host port 8765)
-#        ./deploy.sh staging  — rebuild and restart nexus-staging (host port 8766)
-#
-# Both containers internally serve on 8765; docker-compose.yml maps host ports.
+# Usage: ./deploy.sh          — rebuild and restart nexus-main (port 8765)
+#        ./deploy.sh staging  — rebuild and restart nexus-staging (port 8766)
 set -euo pipefail
 
 SERVICE="${1:-nexus-main}"

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()