Compare commits

...

3 Commits

Author SHA1 Message Date
shokollm
990bc46477 feat(kugetsu): add lock mechanism for worktree coordination
Add lock mechanism to prevent concurrent access to worktrees:
- Add LOCKS_DIR constant (~/.kugetsu/locks)
- Add acquire_lock() - acquires lock file with PID, session_id, timestamp
- Add release_lock() - releases lock if held by current process
- Add release_all_locks() - releases all locks for a session
- Add check_lock() - check if issue is locked
- Modify cmd_start to acquire lock before creating worktree
- Modify cmd_destroy to release lock when destroying session
- Add trap for cleanup on EXIT/INT/TERM

Lock files named after issue ref. Stale lock detection: if PID
no longer exists, lock is considered stale.

Fixes #71
2026-04-02 02:52:16 +00:00
e1050cb70a Merge pull request 'feat(kugetsu): add queue infrastructure for autonomous PM' (#97) from feat/issue-49-queue-v2 into main 2026-04-02 04:37:41 +02:00
shokollm
0f66de2929 feat(kugetsu): add queue infrastructure for autonomous PM
Add queue management system for task queue architecture (Phase 1):

- Add QUEUE_FILE and POLL_INTERVAL constants
- Add init_queue() to initialize queue.json if missing
- Add cmd_queue with subcommands:
  - list: Show queue status with counts per tier
  - enqueue <tier> <msg>: Add task to queue
  - dequeue [tier]: Remove and return next task (priority order)
  - clear: Clear all queued tasks

Queue tiers:
- dev_followups (highest priority)
- user_interrupts (medium)
- background (lowest)

This enables the autonomous queue-based architecture where PM agent
continuously polls the queue and assigns work to dev agents.

Part of #49
2026-04-02 01:04:10 +00:00

View File

@@ -9,7 +9,10 @@ INDEX_FILE="$KUGETSU_DIR/index.json"
NOTIFICATIONS_FILE="$KUGETSU_DIR/notifications.json"
LOGS_DIR="$KUGETSU_DIR/logs"
ENV_DIR="${ENV_DIR:-$KUGETSU_DIR/env}"
QUEUE_FILE="$KUGETSU_DIR/queue.json"
LOCKS_DIR="$KUGETSU_DIR/locks"
MAX_CONCURRENT_AGENTS="${MAX_CONCURRENT_AGENTS:-3}"
POLL_INTERVAL="${POLL_INTERVAL:-600}"
# Load user config overrides (~/.kugetsu/config)
if [ -f "$KUGETSU_DIR/config" ]; then
@@ -56,6 +59,87 @@ count_active_dev_sessions() {
echo "$count"
}
acquire_lock() {
local issue_ref="$1"
local session_id="$2"
local lock_file="$LOCKS_DIR/${issue_ref//[^a-zA-Z0-9._-]/_}.lock"
mkdir -p "$LOCKS_DIR"
if [ -f "$lock_file" ]; then
local lock_pid=$(cat "$lock_file" 2>/dev/null | cut -d: -f1)
local lock_session=$(cat "$lock_file" 2>/dev/null | cut -d: -f2)
if [ -n "$lock_pid" ] && ! kill -0 "$lock_pid" 2>/dev/null; then
echo "Stale lock detected, removing..."
rm -f "$lock_file"
elif [ "$lock_session" = "$session_id" ]; then
echo "Already holding lock for $issue_ref"
return 0
else
echo "Error: $issue_ref is locked by session $lock_session (PID $lock_pid)" >&2
return 1
fi
fi
echo "${BASHPID}:${session_id}:$(date +%s)" > "$lock_file"
echo "Lock acquired for $issue_ref: $lock_file"
return 0
}
release_lock() {
local issue_ref="$1"
local lock_file="$LOCKS_DIR/${issue_ref//[^a-zA-Z0-9._-]/_}.lock"
if [ ! -f "$lock_file" ]; then
return 0
fi
local lock_pid=$(cat "$lock_file" 2>/dev/null | cut -d: -f1)
if [ "$lock_pid" = "$BASHPID" ]; then
rm -f "$lock_file"
echo "Lock released for $issue_ref"
else
echo "Error: Cannot release lock held by PID $lock_pid (current: $BASHPID)" >&2
return 1
fi
}
release_all_locks() {
local session_id="$1"
if [ ! -d "$LOCKS_DIR" ]; then
return 0
fi
for lock_file in "$LOCKS_DIR"/*.lock; do
[ -f "$lock_file" ] || continue
local lock_session=$(cat "$lock_file" 2>/dev/null | cut -d: -f2)
if [ "$lock_session" = "$session_id" ]; then
rm -f "$lock_file"
echo "Released stale lock: $(basename "$lock_file")"
fi
done
}
check_lock() {
local issue_ref="$1"
local lock_file="$LOCKS_DIR/${issue_ref//[^a-zA-Z0-9._-]/_}.lock"
if [ -f "$lock_file" ]; then
local lock_pid=$(cat "$lock_file" 2>/dev/null | cut -d: -f1)
local lock_session=$(cat "$lock_file" 2>/dev/null | cut -d: -f2)
if [ -n "$lock_pid" ] && ! kill -0 "$lock_pid" 2>/dev/null; then
echo "Stale lock detected, removing..."
rm -f "$lock_file"
return 1
fi
echo "Locked by session $lock_session (PID $lock_pid)"
return 0
fi
return 1
}
usage() {
cat << 'EOF'
kugetsu - OpenCode Session Manager (Issue-Driven)
@@ -595,6 +679,116 @@ cmd_logs() {
done
}
init_queue() {
if [ ! -f "$QUEUE_FILE" ]; then
cat > "$QUEUE_FILE" << 'EOF'
{
"dev_followups": [],
"user_interrupts": [],
"background": []
}
EOF
fi
}
cmd_queue() {
local action="${1:-}"
init_queue
case "$action" in
""|"list")
local total=0
echo "Queue status:"
for tier in dev_followups user_interrupts background; do
local count=$(python3 -c "import json; d=json.load(open('$QUEUE_FILE')); print(len(d.get('$tier', [])))" 2>/dev/null || echo 0)
total=$((total + count))
if [ "$count" -eq 0 ]; then
echo " $tier (0): (empty)"
else
echo " $tier ($count):"
python3 -c "import json, sys; d=json.load(open('$QUEUE_FILE')); [print(f' [{t[\"id\"]}] {t[\"message\"][:60]}') for t in d.get('$tier', [])]" 2>/dev/null || echo " (error reading)"
fi
done
echo "Total queued: $total"
;;
"enqueue")
local tier="${2:-}"
local message="${3:-}"
if [ -z "$tier" ] || [ -z "$message" ]; then
echo "Usage: kugetsu queue enqueue <tier> <message>" >&2
echo " tier: dev_followups, user_interrupts, or background" >&2
exit 1
fi
if [[ ! "$tier" =~ ^(dev_followups|user_interrupts|background)$ ]]; then
echo "Error: Invalid tier '$tier'" >&2
echo " Valid tiers: dev_followups, user_interrupts, background" >&2
exit 1
fi
local id="qe-$(date +%s)-$$"
python3 << EOF
import json
with open('$QUEUE_FILE', 'r') as f:
d = json.load(f)
d.setdefault('$tier', []).append({
'id': '$id',
'message': '$message',
'created': '$(date -Iseconds)'
})
with open('$QUEUE_FILE', 'w') as f:
json.dump(d, f, indent=2)
print('Enqueued to $tier: [$id] $message')
EOF
;;
"dequeue")
local tier="${2:-}"
local result=$(python3 << EOF
import json
with open('$QUEUE_FILE', 'r') as f:
d = json.load(f)
tiers = ['dev_followups', 'user_interrupts', 'background'] if not '$tier' else ['$tier']
for t in tiers:
if d.get(t) and len(d[t]) > 0:
task = d[t].pop(0)
with open('$QUEUE_FILE', 'w') as f:
json.dump(d, f, indent=2)
print(f'{t}|{task["id"]}|{task["message"]}')
break
else:
print('Queue empty')
EOF
)
if [ "$result" = "Queue empty" ]; then
echo "$result"
exit 1
fi
echo "$result"
;;
"clear")
cat > "$QUEUE_FILE" << 'EOF'
{
"dev_followups": [],
"user_interrupts": [],
"background": []
}
EOF
echo "Queue cleared"
;;
*)
echo "Usage: kugetsu queue <list|enqueue|dequeue|clear>" >&2
echo "" >&2
echo "Commands:" >&2
echo " list Show queue status" >&2
echo " enqueue <tier> <msg> Add task to queue" >&2
echo " dequeue [tier] Remove and return next task" >&2
echo " clear Clear all queued tasks" >&2
echo "" >&2
echo "Tiers: dev_followups, user_interrupts, background" >&2
exit 1
;;
esac
}
cmd_env() {
local action="${1:-}"
local agent_type="${2:-}"
@@ -1151,6 +1345,14 @@ cmd_start() {
fi
local worktree_path=$(issue_ref_to_worktree_path "$issue_ref")
trap 'release_lock "$issue_ref" 2>/dev/null; exit' EXIT INT TERM
if ! acquire_lock "$issue_ref" "$base_session_id"; then
echo "Error: Could not acquire lock for '$issue_ref'" >&2
exit 1
fi
create_worktree "$issue_ref"
local session_file="$(issue_ref_to_filename "$issue_ref").json"
@@ -1217,6 +1419,8 @@ cmd_start() {
echo "Session started for '$issue_ref': $new_session_id"
echo "Worktree: $worktree_path"
release_lock "$issue_ref"
}
cmd_continue() {
@@ -1470,6 +1674,7 @@ cmd_destroy() {
if [ "$force" = true ]; then
remove_worktree_for_issue "$target"
release_lock "$target" 2>/dev/null || true
rm -f "$session_path"
remove_issue_from_index "$target"
echo "Session for '$target' destroyed"
@@ -1479,6 +1684,7 @@ cmd_destroy() {
read reply
if [ "$reply" = "y" ] || [ "$reply" = "Y" ]; then
remove_worktree_for_issue "$target"
release_lock "$target" 2>/dev/null || true
rm -f "$session_path"
remove_issue_from_index "$target"
echo "Session for '$target' destroyed"
@@ -1523,6 +1729,9 @@ main() {
server)
cmd_server "$@"
;;
queue)
cmd_queue "$@"
;;
env)
cmd_env "$@"
;;