Merge pull request 'fix: queue daemon crashes on every task (issue #174)' (#175) from fix/issue-174-queue-daemon-crashes into main

Fix 3 bugs from issue #174 that caused silent failure loop:

1. kugetsu-log.sh: Fix json.loads with newlines
   - Previously, notifications JSON was embedded in a single-quoted Python
     string literal, but newlines in the JSON broke the Python parser.
   - Fix: Pass JSON via stdin to Python instead of embedding in string.

2. kugetsu-queue-daemon.sh: Create logs directory during init
   - The logs/ directory ($LOGS_DIR) was never created during kugetsu init.
   - Fix: Add mkdir -p for LOGS_DIR, WORKTREES_DIR, QUEUE_DIR, and
     QUEUE_ITEMS_DIR to ensure_dirs() and ensure_queue_dirs().

3. kugetsu: Fix parse_issue_ref_from_message URL parsing
   - The function used buggy grep/sed to parse URLs like
     #158
   - Fix: Use bash regex (=~) for reliable URL parsing with proper
     capture groups.

Additional improvements:
   - ensure_dirs() now creates all necessary directories instead of just
     SESSIONS_DIR
   - ensure_queue_dirs() now also creates QUEUE_DIR and LOGS_DIR
   - parse_issue_ref_from_message uses consistent bash regex approach
     for all URL patterns

docs/CHANGELOG.md

@@ -6,6 +6,25 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ## [Unreleased]
 
+## [v0.2.4] - 2026-04-06
+
+### Fixed
+- Queue daemon: Locking to prevent daemon vs manual conflicts
+- Queue daemon: Proper error handling for failed tasks
+- Queue daemon: Fix GITEA_TOKEN loading from pm-agent.env
+- cmd_delegate: Enqueue tasks instead of bypassing queue
+- Notifications: Call kugetsu_add_notification from bash instead of os.system()
+- kugetsu: Remove duplicate update_queue_item_state that overwrote fixed version
+
+### Added
+- Queue functions moved to kugetsu-index.sh for daemon access
+- kugetsu-session.sh sources required modules for daemon use
+
+## [v0.2.3] - 2026-04-06
+
+### Fixed
+- get_pending_tasks() returns proper JSON array instead of concatenated JSON objects
+
 ## [v0.2.1] - 2026-04-03
 
 ### Fixed

skills/kugetsu/scripts/kugetsu

@@ -93,6 +93,10 @@ EOF
 
 ensure_dirs() {
     mkdir -p "$SESSIONS_DIR"
+    mkdir -p "$LOGS_DIR"
+    mkdir -p "$WORKTREES_DIR"
+    mkdir -p "$QUEUE_DIR"
+    mkdir -p "$QUEUE_ITEMS_DIR"
 }
 
 ensure_worktree_dir() {
@@ -257,7 +261,9 @@ PYEOF
 }
 
 ensure_queue_dirs() {
+    mkdir -p "$QUEUE_DIR"
     mkdir -p "$QUEUE_ITEMS_DIR"
+    mkdir -p "$LOGS_DIR"
 }
 
 generate_queue_id() {
@@ -310,12 +316,31 @@ get_pending_tasks() {
         return
     fi
     
-    find "$QUEUE_ITEMS_DIR" -name "*.json" -type f 2>/dev/null | while read -r file; do
+    python3 -c "
-        local state=$(python3 -c "import json; print(json.load(open('$file')).get('state', ''))" 2>/dev/null || echo "")
+import json
-        if [ "$state" = "pending" ]; then
+import os
-            cat "$file"
+import sys
-        fi
+
-    done | head -"$limit"
+queue_dir = os.environ.get('QUEUE_ITEMS_DIR', '')
+limit = int(sys.argv[1]) if len(sys.argv) > 1 else 10
+
+items = []
+if os.path.isdir(queue_dir):
+    for filename in os.listdir(queue_dir):
+        if filename.endswith('.json'):
+            filepath = os.path.join(queue_dir, filename)
+            try:
+                with open(filepath) as f:
+                    data = json.load(f)
+                    if data.get('state') == 'pending':
+                        items.append(data)
+                        if len(items) >= limit:
+                            break
+            except:
+                pass
+
+print(json.dumps(items))
+" "$limit"
 }
 
 get_queue_stats() {
@@ -342,55 +367,6 @@ get_queue_stats() {
     echo "{\"total\": $total, \"pending\": $pending, \"notified\": $notified, \"completed\": $completed, \"error\": $error}"
 }
 
-update_queue_item_state() {
-    local queue_id="$1"
-    local new_state="$2"
-    local session_id="${3:-}"
-    local pid="${4:-}"
-    
-    local item_file="$QUEUE_ITEMS_DIR/${queue_id}.json"
-    if [ ! -f "$item_file" ]; then
-        echo "Error: Queue item not found: $queue_id" >&2
-        return 1
-    fi
-    
-    python3 << PYEOF
-import json
-import os
-from datetime import datetime
-
-item_file = "$item_file"
-new_state = "$new_state"
-session_id = "$session_id"
-pid = "$pid"
-
-with open(item_file, 'r') as f:
-    item = json.load(f)
-
-issue_ref = item.get('issue_ref', '')
-
-item['state'] = new_state
-
-if new_state == "notified":
-    item['notified_at'] = datetime.now().isoformat() + "Z"
-    if session_id:
-        item['opencode_session_id'] = session_id
-    if pid:
-        item['pid'] = int(pid) if pid.isdigit() else None
-elif new_state == "completed":
-    item['completed_at'] = datetime.now().isoformat() + "Z"
-    os.system(f"kugetsu_add_notification 'task_completed' 'Task completed: {issue_ref}' '{issue_ref}'")
-elif new_state == "error":
-    item['error'] = datetime.now().isoformat() + "Z"
-    os.system(f"kugetsu_add_notification 'task_error' 'Task error: {issue_ref}' '{issue_ref}'")
-
-with open(item_file, 'w') as f:
-    json.dump(item, f, indent=2)
-
-print(f"Updated $queue_id to state: $new_state")
-PYEOF
-}
-
 check_task_timeouts() {
     if [ ! -d "$QUEUE_ITEMS_DIR" ]; then
         return
@@ -878,6 +854,11 @@ EOF
 }
 
 parse_issue_ref_from_message() {
+    # DEPRECATED: This function is not called anywhere.
+    # The active implementation is extract_issue_ref_from_message()
+    # in kugetsu-session.sh which is used by cmd_delegate.
+    # This function is kept for backwards compatibility and will
+    # be removed in a future release.
     local message="$1"
     
     local gitserver=""
@@ -885,21 +866,20 @@ parse_issue_ref_from_message() {
     local repo=""
     local issue_number=""
     
-    if echo "$message" | grep -qE '[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+/(issues|pull)/[0-9]+'; then
+    if [[ "$message" =~ (https?://)?([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})/([a-zA-Z0-9._-]+)/([a-zA-Z0-9._-]+)/(issues|pull)/([0-9]+) ]]; then
-        gitserver=$(echo "$message" | grep -oE '[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+' | head -1 | sed 's/\/[^/]*\/[^/]*$//')
+        gitserver="${BASH_REMATCH[2]}"
-        local full_path=$(echo "$message" | grep -oE '[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+/(issues|pull)/[0-9]+' | head -1)
+        owner="${BASH_REMATCH[3]}"
-        owner=$(echo "$full_path" | cut -d'/' -f2)
+        repo="${BASH_REMATCH[4]}"
-        repo=$(echo "$full_path" | cut -d'/' -f3)
+        issue_number="${BASH_REMATCH[6]}"
-        issue_number=$(echo "$full_path" | grep -oE '[0-9]+$' | head -1)
+    elif [[ "$message" =~ (https?://)?([a-zA-Z0-9.-]+\.[a-zA-Z]{2,})/([a-zA-Z0-9._-]+)/([a-zA-Z0-9._-]+)#([0-9]+) ]]; then
-    elif echo "$message" | grep -qE '[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+#[0-9]+'; then
+        gitserver="${BASH_REMATCH[2]}"
-        gitserver=$(echo "$message" | grep -oE '[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}/[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+' | head -1)
+        owner="${BASH_REMATCH[3]}"
-        owner=$(echo "$gitserver" | cut -d'/' -f2)
+        repo="${BASH_REMATCH[4]}"
-        repo=$(echo "$gitserver" | cut -d'/' -f3)
+        issue_number="${BASH_REMATCH[5]}"
-        issue_number=$(echo "$message" | grep -oE '#[0-9]+' | grep -oE '[0-9]+' | head -1)
+    elif [[ "$message" =~ ([a-zA-Z0-9._-]+)/([a-zA-Z0-9._-]+)#([0-9]+) ]]; then
-    elif echo "$message" | grep -qE '[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+#([0-9]+)'; then
+        owner="${BASH_REMATCH[1]}"
-        owner=$(echo "$message" | grep -oE '[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+#' | sed 's/#$//' | cut -d'/' -f1)
+        repo="${BASH_REMATCH[2]}"
-        repo=$(echo "$message" | grep -oE '[a-zA-Z0-9_.-]+/[a-zA-Z0-9_.-]+#' | sed 's/#$//' | cut -d'/' -f2)
+        issue_number="${BASH_REMATCH[3]}"
-        issue_number=$(echo "$message" | grep -oE '#[0-9]+' | grep -oE '[0-9]+' | head -1)
     fi
     
     echo "${gitserver}|${owner}|${repo}|${issue_number}"

skills/kugetsu/scripts/kugetsu-index.sh

@@ -202,9 +202,10 @@ update_queue_item_state() {
         return 1
     fi
     
+    local issue_ref=$(python3 -c "import json; print(json.load(open('$item_file')).get('issue_ref', ''))" 2>/dev/null || echo "")
+    
     python3 << PYEOF
 import json
-import os
 from datetime import datetime
 
 item_file = "$item_file"
@@ -215,8 +216,6 @@ pid = "$pid"
 with open(item_file, 'r') as f:
     item = json.load(f)
 
-issue_ref = item.get('issue_ref', '')
-
 item['state'] = new_state
 
 if new_state == "notified":
@@ -227,14 +226,18 @@ if new_state == "notified":
         item['pid'] = int(pid) if pid.isdigit() else None
 elif new_state == "completed":
     item['completed_at'] = datetime.now().isoformat() + "Z"
-    os.system(f"kugetsu_add_notification 'task_completed' 'Task completed: {issue_ref}' '{issue_ref}'")
 elif new_state == "error":
     item['error'] = datetime.now().isoformat() + "Z"
-    os.system(f"kugetsu_add_notification 'task_error' 'Task error: {issue_ref}' '{issue_ref}'")
 
 with open(item_file, 'w') as f:
     json.dump(item, f, indent=2)
 
 print(f"Updated $queue_id to state: $new_state")
 PYEOF
+    
+    if [ "$new_state" = "completed" ]; then
+        kugetsu_add_notification "task_completed" "Task completed: $issue_ref" "$issue_ref"
+    elif [ "$new_state" = "error" ]; then
+        kugetsu_add_notification "task_error" "Task error: $issue_ref" "$issue_ref"
+    fi
 }

skills/kugetsu/scripts/kugetsu-log.sh

@@ -43,15 +43,24 @@ kugetsu_add_notification() {
         notifications=$(cat "$NOTIFICATIONS_FILE")
     fi
     
-    local new_notification=$(python3 -c "import json; print(json.dumps({
+    notifications=$(echo "$notifications" | python3 -c "
-        'type': '$notification_type',
+import json
-        'message': '$message',
+import sys
-        'issue_ref': '$issue_ref',
-        'timestamp': '$timestamp',
-        'read': False
-    }))")
 
-    notifications=$(python3 -c "import json; n=json.loads('$notifications'); n.append(json.loads('$new_notification')); print(json.dumps(n[-50:] if len(n)>50 else n, indent=2))")
+notifications = json.load(sys.stdin)
+new_notification = {
+    'type': '$notification_type',
+    'message': '''$message'''.replace('\"', '\"'),
+    'issue_ref': '$issue_ref' if '$issue_ref' else None,
+    'timestamp': '$timestamp',
+    'read': False
+}
+
+notifications.append(new_notification)
+notifications = notifications[-50:] if len(notifications) > 50 else notifications
+
+print(json.dumps(notifications, indent=2))
+")
     
     echo "$notifications" > "$NOTIFICATIONS_FILE"
 }

skills/kugetsu/scripts/kugetsu-queue-daemon.sh

@@ -8,12 +8,29 @@ source "$SCRIPT_DIR/kugetsu-index.sh"
 source "$SCRIPT_DIR/kugetsu-worktree.sh"
 source "$SCRIPT_DIR/kugetsu-log.sh"
 
-# Load GITEA_TOKEN from default.env
+load_agent_env "pm-agent"
-if [ -f "$HOME/.kugetsu/env/default.env" ]; then
+
-    source "$HOME/.kugetsu/env/default.env"
+acquire_lock() {
-fi
+    local issue_ref="$1"
+    local lock_file="$QUEUE_DIR/locks/$(echo "$issue_ref" | sed 's/[\/:]/-/g' | sed 's/#/-/').lock"
+    mkdir -p "$(dirname "$lock_file")"
+    if [ -f "$lock_file" ]; then
+        local pid=$(cat "$lock_file" 2>/dev/null || echo "")
+        if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then
+            return 1
+        fi
+        rm -f "$lock_file"
+    fi
+    echo $$ > "$lock_file"
+    return 0
+}
+
+release_lock() {
+    local issue_ref="$1"
+    local lock_file="$QUEUE_DIR/locks/$(echo "$issue_ref" | sed 's/[\/:]/-/g' | sed 's/#/-/').lock"
+    rm -f "$lock_file"
+}
 
-# Check if a notified task has completed (forked session ended or has new commits)
 check_task_completion() {
     local item="$1"
     local queue_id=$(basename "$item" .json)
@@ -21,71 +38,116 @@ check_task_completion() {
 
     [ "$state" = "notified" ] || return 0
 
-    # Use opencode_session_id (the forked session, not the parent pm_session)
     local session_id=$(python3 -c "import json; print(json.load(open('$item')).get('opencode_session_id', ''))" 2>/dev/null)
     local issue_ref=$(python3 -c "import json; print(json.load(open('$item')).get('issue_ref', ''))" 2>/dev/null)
+    local pid=$(python3 -c "import json; print(json.load(open('$item')).get('pid', ''))" 2>/dev/null)
 
-    # If no session tracked, skip
+    if [ -n "$pid" ] && [ "$pid" != "None" ]; then
-    [ -n "$session_id" ] || return 0
+        if ! kill -0 "$pid" 2>/dev/null; then
+            local worktree_path=$(issue_ref_to_worktree_path "$issue_ref" "$HOME/.kugetsu-worktrees")
+            local has_commits=false
 
-    # Check if forked session still exists in opencode
+            if [ -d "$worktree_path" ] && [ -d "$worktree_path/.git" ]; then
-    if ! opencode session list 2>/dev/null | grep -q "$session_id"; then
+                if [ -n "$(git -C "$worktree_path" log --oneline origin/main..HEAD 2>/dev/null)" ]; then
-        # Forked session ended — check if work was done
+                    has_commits=true
-        local worktree_path=$(issue_ref_to_worktree_path "$issue_ref" "$HOME/.kugetsu-worktrees")
+                fi
-        local has_commits=false
-        
-        if [ -d "$worktree_path" ] && [ -d "$worktree_path/.git" ]; then
-            # Check if worktree has new commits beyond origin/main
-            if [ -n "$(git -C "$worktree_path" log --oneline origin/main..HEAD 2>/dev/null)" ]; then
-                has_commits=true
             fi
-        fi
 
-        if [ "$has_commits" = true ]; then
+            if [ "$has_commits" = true ]; then
-            update_queue_item_state "$queue_id" "completed"
+                update_queue_item_state "$queue_id" "completed"
-            echo "Task $queue_id ($issue_ref) completed — new commits found"
+                echo "Task $queue_id ($issue_ref) completed — new commits found"
-        else
+            else
-            update_queue_item_state "$queue_id" "error"
+                update_queue_item_state "$queue_id" "error"
-            echo "Task $queue_id ($issue_ref) marked error — no commits found after session ended"
+                echo "Task $queue_id ($issue_ref) marked error — no commits found after session ended"
+            fi
+            release_lock "$issue_ref"
+        fi
+    else
+        if [ -n "$session_id" ] && ! opencode session list 2>/dev/null | grep -q "$session_id"; then
+            local worktree_path=$(issue_ref_to_worktree_path "$issue_ref" "$HOME/.kugetsu-worktrees")
+            local has_commits=false
+
+            if [ -d "$worktree_path" ] && [ -d "$worktree_path/.git" ]; then
+                if [ -n "$(git -C "$worktree_path" log --oneline origin/main..HEAD 2>/dev/null)" ]; then
+                    has_commits=true
+                fi
+            fi
+
+            if [ "$has_commits" = true ]; then
+                update_queue_item_state "$queue_id" "completed"
+                echo "Task $queue_id ($issue_ref) completed — new commits found"
+            else
+                update_queue_item_state "$queue_id" "error"
+                echo "Task $queue_id ($issue_ref) marked error — no commits found after session ended"
+            fi
+            release_lock "$issue_ref"
         fi
     fi
 }
 
+get_session_id_for_issue() {
+    local issue_ref="$1"
+    local session_file=$(issue_ref_to_filename "$issue_ref")
+    local session_path="$SESSIONS_DIR/$session_file"
+    if [ -f "$session_path" ]; then
+        python3 -c "import json; print(json.load(open('$session_path')).get('opencode_session_id', ''))" 2>/dev/null || echo ""
+    else
+        echo ""
+    fi
+}
+
+process_task() {
+    local item="$1"
+    local queue_id=$(basename "$item" .json)
+    local issue_ref=$(python3 -c "import json; print(json.load(open('$item')).get('issue_ref', ''))" 2>/dev/null)
+    local message=$(python3 -c "import json; print(json.load(open('$item')).get('message', ''))" 2>/dev/null)
+
+    if ! acquire_lock "$issue_ref"; then
+        echo "Task $queue_id ($issue_ref) skipped — another process is handling it"
+        return
+    fi
+
+    source "$SCRIPT_DIR/kugetsu-session.sh"
+
+    if worktree_exists "$issue_ref" "$HOME/.kugetsu-worktrees" || [ -f "$SESSIONS_DIR/$(issue_ref_to_filename "$issue_ref").json" ]; then
+        log_file="$LOGS_DIR/delegate-$(date +%s).log"
+        if cmd_continue "$issue_ref" "$message" >> "$log_file" 2>&1; then
+            sleep 1
+            local session_id=$(get_session_id_for_issue "$issue_ref")
+            update_queue_item_state "$queue_id" "notified" "$session_id" ""
+            echo "Task $queue_id continued for $issue_ref"
+        else
+            update_queue_item_state "$queue_id" "error"
+            echo "Task $queue_id ($issue_ref) failed to continue"
+        fi
+    else
+        log_file="$LOGS_DIR/delegate-$(date +%s).log"
+        if cmd_start "$issue_ref" "$message" >> "$log_file" 2>&1; then
+            sleep 1
+            local session_id=$(get_session_id_for_issue "$issue_ref")
+            update_queue_item_state "$queue_id" "notified" "$session_id" ""
+            echo "Task $queue_id started for $issue_ref"
+        else
+            update_queue_item_state "$queue_id" "error"
+            echo "Task $queue_id ($issue_ref) failed to start"
+        fi
+    fi
+
+    release_lock "$issue_ref"
+}
+
 while true; do
-    # Check completion of notified tasks
     if [ -d "$QUEUE_ITEMS_DIR" ]; then
         for item in "$QUEUE_ITEMS_DIR"/*.json; do
             [ -f "$item" ] || continue
             check_task_completion "$item"
         done
 
-        # Process pending tasks
         for item in "$QUEUE_ITEMS_DIR"/*.json; do
             [ -f "$item" ] || continue
             state=$(python3 -c "import json; print(json.load(open('$item')).get('state', ''))" 2>/dev/null)
             if [ "$state" = "pending" ]; then
-                queue_id=$(basename "$item" .json)
+                process_task "$item"
-                issue_ref=$(python3 -c "import json; print(json.load(open('$item')).get('issue_ref', ''))" 2>/dev/null)
-                message=$(python3 -c "import json; print(json.load(open('$item')).get('message', ''))" 2>/dev/null)
-                
-                # Source session management and use cmd_start/cmd_continue
-                source "$SCRIPT_DIR/kugetsu-session.sh"
-                
-                if worktree_exists "$issue_ref" "$HOME/.kugetsu-worktrees" || [ -f "$SESSIONS_DIR/$(issue_ref_to_filename "$issue_ref").json" ]; then
-                    # Continue existing session
-                    log_file="$LOGS_DIR/delegate-$(date +%s).log"
-                    cmd_continue "$issue_ref" "$message" >> "$log_file" 2>&1 &
-                    pid=$!
-                    update_queue_item_state "$queue_id" "notified" "" "$pid"
-                    echo "Task $queue_id continued for $issue_ref"
-                else
-                    # Start new session
-                    log_file="$LOGS_DIR/delegate-$(date +%s).log"
-                    cmd_start "$issue_ref" "$message" >> "$log_file" 2>&1 &
-                    pid=$!
-                    update_queue_item_state "$queue_id" "notified" "" "$pid"
-                    echo "Task $queue_id started for $issue_ref"
-                fi
             fi
         done
     fi

skills/kugetsu/scripts/kugetsu-session.sh

@@ -81,9 +81,20 @@ EOF
     echo "Press Ctrl+C to cancel or wait for session to be created"
     sleep 2
 
+    local before_sessions=$(opencode session list 2>/dev/null | grep -E '^ses_' | awk '{print $1}' || true)
+
     opencode
 
-    local session_ids=$(opencode session list 2>/dev/null | grep -E '^ses_' | awk '{print $1}' | tail -1)
+    local after_sessions=$(opencode session list 2>/dev/null | grep -E '^ses_' | awk '{print $1}' || true)
+    local session_ids=""
+    while IFS= read -r line; do
+        local sid=$(echo "$line" | awk '{print $1}')
+        if [ -n "$sid" ] && ! echo "$before_sessions" | grep -q "^${sid}$"; then
+            session_ids="$sid"
+            break
+        fi
+    done <<< "$after_sessions"
+
     if [ -z "$session_ids" ]; then
         echo "Error: Could not find newly created session" >&2
         exit 1
@@ -95,9 +106,20 @@ EOF
     echo "Base session created: $session_ids"
     echo "Starting PM agent..."
 
+    before_sessions="$after_sessions"
+
     opencode
 
-    local pm_session_ids=$(opencode session list 2>/dev/null | grep -E '^ses_' | grep -v "$session_ids" | tail -1)
+    after_sessions=$(opencode session list 2>/dev/null | grep -E '^ses_' | awk '{print $1}' || true)
+    local pm_session_ids=""
+    while IFS= read -r line; do
+        local sid=$(echo "$line" | awk '{print $1}')
+        if [ -n "$sid" ] && ! echo "$before_sessions" | grep -q "^${sid}$"; then
+            pm_session_ids="$sid"
+            break
+        fi
+    done <<< "$after_sessions"
+
     if [ -z "$pm_session_ids" ]; then
         echo "Warning: Could not find separate PM agent session" >&2
         pm_session_ids="$session_ids"