Merge pull request 'fix: init script captures wrong session IDs when old sessions exist' (#173) from fix/issue-172-init-script-wrong-session-ids into main

The init script used 'tail -1' to find newly created sessions, which
fails when old sessions exist because it picks an existing session
instead of the newly created one.

The fix captures session IDs before and after creating new sessions,
then diffs to identify newly created sessions.

Fixes #172

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

@@ -310,12 +310,31 @@ get_pending_tasks() {
         return
     fi
     
-    find "$QUEUE_ITEMS_DIR" -name "*.json" -type f 2>/dev/null | while read -r file; do
-        local state=$(python3 -c "import json; print(json.load(open('$file')).get('state', ''))" 2>/dev/null || echo "")
-        if [ "$state" = "pending" ]; then
-            cat "$file"
-        fi
-    done | head -"$limit"
+    python3 -c "
+import json
+import os
+import sys
+
+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 +361,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

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-queue-daemon.sh

@@ -8,86 +8,148 @@ source "$SCRIPT_DIR/kugetsu-index.sh"
 source "$SCRIPT_DIR/kugetsu-worktree.sh"
 source "$SCRIPT_DIR/kugetsu-log.sh"
 
-# Load GITEA_TOKEN from default.env
-if [ -f "$HOME/.kugetsu/env/default.env" ]; then
-    source "$HOME/.kugetsu/env/default.env"
-fi
+load_agent_env "pm-agent"
+
+acquire_lock() {
+    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)
     local state=$(python3 -c "import json; print(json.load(open('$item')).get('state', ''))" 2>/dev/null)
-    
+
     [ "$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)
-    
-    # If no session tracked, skip
-    [ -n "$session_id" ] || return 0
-    
-    # Check if forked session still exists in opencode
-    if ! opencode session list 2>/dev/null | grep -q "$session_id"; then
-        # Forked session ended — check if work was done
-        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
-            # 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
+    local pid=$(python3 -c "import json; print(json.load(open('$item')).get('pid', ''))" 2>/dev/null)
+
+    if [ -n "$pid" ] && [ "$pid" != "None" ]; then
+        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
+
+            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
-        
-        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"
+    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)
-                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
+                process_task "$item"
             fi
         done
     fi
     sleep "${QUEUE_DAEMON_INTERVAL_MINUTES:-5}m"
-done
+done

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"