Architecture: Enable concurrent agent sessions per repository

[tbrandenburg]

Problem

The current channel system uses repository-level locking instead of session-based locking, preventing multiple users from working simultaneously on the same repository.

Current Behavior

Channel Identifier: channel = repository_name (from packages/pybackend/app.py:502)

# Only ONE agent can run per repository across ALL sessions
def _mark_channel_processing(channel: str) -> bool:
    with _processing_lock:
        if channel in _processing_channels:  # Global lock per repo
            return False  # HTTP 409 Conflict to ALL sessions

Impact on Multi-User Workflows

Scenario:

👤 Developer A: Working on "my-project" → Starts agent → Channel locked
👤 Developer B: Working on "my-project" → Gets HTTP 409 Conflict  
👤 Developer C: Reviewing "my-project" → Also blocked
⏱️  3 minutes later: Developer A completes → Everyone else can proceed

Problems:

• 🚫 No concurrent development on popular repositories
• 🚫 Demo failures when multiple people access the same project
• 🚫 Poor team collaboration experience with cryptic 409 errors

Proposed Solution

Session-Based Channel System

Change from:

channel = name  # Repository-wide lock (current)

Change to:

channel = f"{name}:{session_id}"  # Per-session lock (proposed)

Benefits

• ✅ Multiple users can work simultaneously on same repository
• ✅ Session isolation prevents cross-talk between concurrent runs
• ✅ Better team collaboration experience
• ✅ Eliminates demo/presentation failures

Technical Considerations

Files to Modify:

• packages/pybackend/agent_service.py - Channel management system
• packages/pybackend/app.py - All agent endpoint handlers
• Tests for concurrent session behavior

Risks to Address:

• Workspace file conflicts during concurrent operations
• Resource contention (CPU, memory, file locks)
• Data consistency between concurrent agent runs
• Proper cleanup of session-based channels

Acceptance Criteria

• Multiple users can run agents simultaneously on same repository
• Session isolation prevents cross-talk between runs
• Workspace file conflicts handled gracefully (locking or merge strategies)
• Performance remains acceptable with concurrent operations
• Updated API documentation for new concurrency model
• Comprehensive test coverage for concurrent scenarios

Related Issues

• Bug: Chat client re-sends user messages after receiving a valid response (retry loop) #358 - Chat client retry loop (separate frontend issue)

Priority

Medium - Improves team workflows but current single-session behavior is acceptable for now.

Epic

Multi-user concurrent development support

[github-actions]

Hello tbrandenburg!

Thanks for your feedback!

Here are the tools I am offering:

Commands:
/ghar-commit-push
/ghar-issue-fix
/ghar-issue-investigate
/ghar-maintainability-review
/ghar-resolve-ci-errors
/ghar-review
/ghar-security-check
/ghar-stale-check

Agents:
*review

Usage:

• Use commands directly: /command-name
• Use agents with commands: *agent-name /command-name PROMPT

Example: *review /security-review Please check this code for vulnerabilities

[github-actions]

Investigation: Architecture: Enable concurrent agent sessions per repository

Issue: #360 (#360)
Type: ENHANCEMENT
Investigated: 2026-04-03T00:00:00Z

Assessment

Metric	Value	Reasoning
Priority	MEDIUM	Improves team workflows but current single-session behavior is acceptable; blocking for multi-user demos and team use
Complexity	MEDIUM	2 files to modify (app.py, agent_service.py), but requires careful handling of 4 different agent endpoint types
Confidence	HIGH	Clear root cause identified, session_id already available in API layer, working directory extraction is straightforward

---

Problem Statement

The current channel system uses repository-level locking (channel = repository_name), preventing multiple users from working simultaneously on the same repository. When one user starts an agent, all other users on the same repository receive HTTP 409 Conflict errors until the first user completes.

---

Analysis

Change Rationale

The session_id is already available in all API endpoints and passed to send_agent_message, but it is only used to track conversations AFTER the channel lock is acquired. By incorporating session_id into the channel identifier itself, each session can have its own independent lock while still sharing the repository's working directory.

Current Flow:

1. API receives request with session_id (available but unused for locking)
2. send_agent_message(name, ...) called with name as channel
3. _mark_channel_processing(name) blocks if any session is using name
4. Only AFTER lock acquired: _conversation_sessions[channel] = session_id

Proposed Flow:

1. API receives request with session_id
2. send_agent_message(f"{name}:{session_id}", ...) called with composite channel
3. Each session gets independent lock via its unique channel
4. _get_working_directory() extracts repo name from channel for shared workspace

Evidence Chain

Current Implementation - Repository Agent (app.py:502):

return send_agent_message(name, message, session_id, agent, model)
# channel = "my-repo" - blocks ALL sessions for "my-repo"

Current Implementation - Channel Locking (agent_service.py:148-153):

def _mark_channel_processing(channel: str) -> bool:
    with _processing_lock:
        if channel in _processing_channels:  # Global per repo name
            return False
        _processing_channels[channel] = datetime.now(UTC)
        return True

Current Implementation - Working Directory (agent_service.py:294-302):

def _get_working_directory(channel: str) -> Path:
    if not channel.startswith("knowledge:") and not channel.startswith("constitution:"):
        workspace = get_workspace_home()
        repo_path = workspace / channel  # Uses full channel as repo path
        if repo_path.exists() and repo_path.is_dir():
            return repo_path

Affected Files

File	Lines	Action	Description
packages/pybackend/app.py	502	UPDATE	Change channel to include session_id for repo agent
packages/pybackend/app.py	509	UPDATE	Add sessionId query param to repository status
packages/pybackend/app.py	514	UPDATE	Add sessionId query param to repository cancel
packages/pybackend/app.py	1009	UPDATE	Change channel for knowledge agent
packages/pybackend/app.py	1017	UPDATE	Add sessionId param to knowledge status
packages/pybackend/app.py	1022	UPDATE	Add sessionId param to knowledge cancel
packages/pybackend/app.py	1218	UPDATE	Change channel for constitution agent
packages/pybackend/app.py	1226	UPDATE	Add sessionId param to constitution status
packages/pybackend/app.py	1231	UPDATE	Add sessionId param to constitution cancel
packages/pybackend/app.py	1360	UPDATE	Change channel for task agent
packages/pybackend/app.py	1367	UPDATE	Add sessionId param to task status
packages/pybackend/app.py	1372	UPDATE	Add sessionId param to task cancel
packages/pybackend/agent_service.py	297-301	UPDATE	Extract repo name from composite channel
packages/pybackend/agent_service.py	NEW	CREATE	Add helper function _get_repo_name_from_channel()
packages/pybackend/tests/unit/test_api.py	505-543	UPDATE	Update tests for session-based channels
packages/pybackend/tests/unit/test_agent_service.py	NEW	CREATE	Add tests for concurrent session behavior

Integration Points

• API Endpoints: 4 agent types (repository, knowledge, constitution, task) all need session-based channels
• Status/Cancel Endpoints: Must accept sessionId parameter to target correct session's channel
• Working Directory: Must extract repo name from composite channel ("my-repo:abc123" → "my-repo")
• Frontend: Already sends sessionId in request body; no frontend changes needed

---

Implementation Plan

Step 1: Add helper function to extract repo name from composite channel

File: packages/pybackend/agent_service.py
Lines: 294 (after imports)
Action: CREATE (new function)

def _get_repo_name_from_channel(channel: str) -> str:
    """Extract the repository/resource name from a composite channel identifier.
    
    Channels have formats:
    - Repository: "repo-name" or "repo-name:session-id"
    - Knowledge: "knowledge:resource-name" or "knowledge:resource-name:session-id"
    - Constitution: "constitution:name" or "constitution:name:session-id"
    - Task: "task:name" or "task:name:session-id"
    """
    parts = channel.rsplit(":", 1)
    if len(parts) == 2 and not _looks_like_resource_prefix(parts[0]):
        return parts[0]
    if channel.startswith("knowledge:") or channel.startswith("constitution:") or channel.startswith("task:"):
        prefix, rest = channel.split(":", 1)
        resource = rest.rsplit(":", 1)[0]
        return f"{prefix}:{resource}"
    return channel


def _looks_like_resource_prefix(prefix: str) -> bool:
    """Check if a string looks like a resource prefix (knowledge:, constitution:, task:)."""
    return prefix in ("knowledge", "constitution", "task")

Why: Centralize channel parsing logic to avoid duplication across functions.

---

Step 2: Update _get_working_directory to use helper

File: packages/pybackend/agent_service.py
Lines: 294-310
Action: UPDATE

Current code:

def _get_working_directory(channel: str) -> Path:
    """Determine the working directory based on the channel context."""
    if not channel.startswith("knowledge:") and not channel.startswith("constitution:"):
        workspace = get_workspace_home()
        repo_path = workspace / channel  # Uses full channel
        if repo_path.exists() and repo_path.is_dir():
            return repo_path
        return Path(__file__).parent

    made_dir = get_made_directory()

    if channel.startswith("knowledge:"):
        return ensure_directory(made_dir / "knowledge")

    return ensure_directory(made_dir / "constitutions")

Required change:

def _get_working_directory(channel: str) -> Path:
    """Determine the working directory based on the channel context."""
    if not channel.startswith("knowledge:") and not channel.startswith("constitution:"):
        workspace = get_workspace_home()
        repo_name = _get_repo_name_from_channel(channel)  # Extract repo name
        repo_path = workspace / repo_name
        if repo_path.exists() and repo_path.is_dir():
            return repo_path
        return Path(__file__).parent

    made_dir = get_made_directory()

    if channel.startswith("knowledge:"):
        return ensure_directory(made_dir / "knowledge")

    return ensure_directory(made_dir / "constitutions")

Why: The function must extract the repository name from composite channels to maintain shared workspace behavior.

---

Step 3: Update repository agent endpoint

File: packages/pybackend/app.py
Lines: 485-519
Action: UPDATE

Current code (lines 485-504):

@app.post("/api/repositories/{name}/agent")
def repository_agent(name: str, payload: dict = Body(...)):
    message = payload.get("message")
    session_id = payload.get("sessionId")
    agent = payload.get("agent")
    model = payload.get("model")
    if not message:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Message is required"
        )
    try:
        logger.info(
            "Forwarding agent message for repository '%s' (session: %s, agent: %s)",
            name,
            session_id or "new",
            agent or "default",
        )
        return send_agent_message(name, message, session_id, agent, model)
    except ChannelBusyError as exc:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))


@app.get("/api/repositories/{name}/agent/status")
def repository_agent_status(name: str):
    return get_channel_status(name)


@app.post("/api/repositories/{name}/agent/cancel")
def repository_agent_cancel(name: str):
    if not cancel_agent_message(name):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No active agent process to cancel",
        )
    return {"success": True}

Required change:

@app.post("/api/repositories/{name}/agent")
def repository_agent(name: str, payload: dict = Body(...)):
    message = payload.get("message")
    session_id = payload.get("sessionId")
    agent = payload.get("agent")
    model = payload.get("model")
    if not message:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="Message is required"
        )
    if not session_id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST, detail="sessionId is required for agent requests"
        )
    channel = f"{name}:{session_id}"
    try:
        logger.info(
            "Forwarding agent message for repository '%s' (session: %s, agent: %s)",
            name,
            session_id,
            agent or "default",
        )
        return send_agent_message(channel, message, session_id, agent, model)
    except ChannelBusyError as exc:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc))


@app.get("/api/repositories/{name}/agent/status")
def repository_agent_status(name: str, sessionId: str = Query(..., description="The session ID")):
    return get_channel_status(f"{name}:{sessionId}")


@app.post("/api/repositories/{name}/agent/cancel")
def repository_agent_cancel(name: str, sessionId: str = Query(..., description="The session ID")):
    if not cancel_agent_message(f"{name}:{sessionId}"):
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="No active agent process to cancel",
        )
    return {"success": True}

Why: Each session gets its own channel for independent locking while preserving repository-level workspace.

---

Step 4: Update knowledge, constitution, and task agent endpoints

Apply the same pattern to:

Lines 991-1024 (knowledge):

# Change line 1008-1009 from:
return send_agent_message(
    f"knowledge:{name}", message, session_id, agent, model
)
# To:
return send_agent_message(
    f"knowledge:{name}:{session_id}", message, session_id, agent, model
)

# Change line 1016-1017 from:
def knowledge_agent_status(name: str):
    return get_channel_status(f"knowledge:{name}")
# To:
def knowledge_agent_status(name: str, sessionId: str = Query(..., description="The session ID")):
    return get_channel_status(f"knowledge:{name}:{sessionId}")

# Change line 1021-1022 from:
def knowledge_agent_cancel(name: str):
    if not cancel_agent_message(f"knowledge:{name}"):
# To:
def knowledge_agent_cancel(name: str, sessionId: str = Query(..., description="The session ID")):
    if not cancel_agent_message(f"knowledge:{name}:{sessionId}"):

Lines 1200-1236 (constitution):

# Change line 1217-1218 from:
return send_agent_message(
    f"constitution:{name}", message, session_id, agent, model
)
# To:
return send_agent_message(
    f"constitution:{name}:{session_id}", message, session_id, agent, model
)

# Update constitution_agent_status and constitution_agent_cancel similarly

Lines 1343-1377 (task):

# Change line 1360 from:
return send_agent_message(f"task:{name}", message, session_id, agent, model)
# To:
return send_agent_message(f"task:{name}:{session_id}", message, session_id, agent, model)

# Update task_agent_status and task_agent_cancel similarly

Why: All agent types must support concurrent sessions with session-based locking.

---

Step 5: Update tests for API endpoints

File: packages/pybackend/tests/unit/test_api.py
Lines: 505-543
Action: UPDATE

Add new tests:

@patch("app.send_agent_message")
def test_repository_agent_concurrent_sessions(self, mock_send):
    """Test that multiple sessions can run concurrently on same repository."""
    mock_send.return_value = {"messageId": "1", "processing": True}
    
    # First session
    response1 = client.post(
        "/api/repositories/sample/agent", 
        json={"message": "hello", "sessionId": "session-1"}
    )
    assert response1.status_code == 200
    mock_send.assert_called_with("sample:session-1", "hello", "session-1", None, None)
    
    # Second session (should NOT conflict)
    response2 = client.post(
        "/api/repositories/sample/agent", 
        json={"message": "hello", "sessionId": "session-2"}
    )
    assert response2.status_code == 200
    mock_send.assert_called_with("sample:session-2", "hello", "session-2", None, None)

@patch("app.send_agent_message")
def test_repository_agent_requires_session_id(self, mock_send):
    """Test that sessionId is required for agent requests."""
    response = client.post(
        "/api/repositories/sample/agent", 
        json={"message": "hello"}
    )
    assert response.status_code == 400
    assert "sessionId" in response.json()["detail"]

@patch("app.get_channel_status")
def test_repository_agent_status_with_session(self, mock_status):
    """Test status endpoint with sessionId."""
    mock_status.return_value = {"processing": True, "startedAt": "2024-01-01T00:00:00Z"}
    
    response = client.get("/api/repositories/sample/agent/status?sessionId=abc123")
    
    assert response.status_code == 200
    mock_status.assert_called_once_with("sample:abc123")

@patch("app.cancel_agent_message")
def test_repository_agent_cancel_with_session(self, mock_cancel):
    """Test cancel endpoint with sessionId."""
    mock_cancel.return_value = True
    
    response = client.post("/api/repositories/sample/agent/cancel?sessionId=abc123")
    
    assert response.status_code == 200
    mock_cancel.assert_called_once_with("sample:abc123")

Why: Ensure concurrent session behavior is tested and sessionId requirement is enforced.

---

Step 6: Add unit tests for concurrent channel behavior

File: packages/pybackend/tests/unit/test_agent_service.py
Action: CREATE

import pytest
from unittest.mock import patch, MagicMock
from agent_service import (
    _mark_channel_processing,
    _clear_channel_processing,
    _get_repo_name_from_channel,
    _get_working_directory,
    send_agent_message,
)


class TestChannelParsing:
    def test_get_repo_name_from_channel_simple(self):
        assert _get_repo_name_from_channel("my-repo") == "my-repo"
    
    def test_get_repo_name_from_channel_with_session(self):
        assert _get_repo_name_from_channel("my-repo:session-123") == "my-repo"
    
    def test_get_repo_name_from_channel_knowledge(self):
        assert _get_repo_name_from_channel("knowledge:docs") == "knowledge:docs"
        assert _get_repo_name_from_channel("knowledge:docs:session-123") == "knowledge:docs"
    
    def test_get_repo_name_from_channel_constitution(self):
        assert _get_repo_name_from_channel("constitution:rules") == "constitution:rules"
        assert _get_repo_name_from_channel("constitution:rules:session-123") == "constitution:rules"


class TestConcurrentSessions:
    def test_multiple_sessions_same_repo(self):
        """Test that multiple sessions can acquire channels for same repo."""
        _clear_channel_processing("repo:session-1")
        _clear_channel_processing("repo:session-2")
        
        # Session 1 should succeed
        assert _mark_channel_processing("repo:session-1") == True
        
        # Session 2 should also succeed (different channel)
        assert _mark_channel_processing("repo:session-2") == True
        
        # Cleanup
        _clear_channel_processing("repo:session-1")
        _clear_channel_processing("repo:session-2")
    
    def test_same_session_blocks(self):
        """Test that same session cannot double-book."""
        _clear_channel_processing("repo:session-1")
        
        assert _mark_channel_processing("repo:session-1") == True
        assert _mark_channel_processing("repo:session-1") == False  # Already locked
        
        _clear_channel_processing("repo:session-1")

Why: Verify core channel locking behavior and parsing logic.

---

Patterns to Follow

From codebase - session_id handling:

# SOURCE: app.py:488
# Pattern for extracting session_id from payload
session_id = payload.get("sessionId")

# SOURCE: app.py:500-501
# Pattern for logging with session context
logger.info(
    "Forwarding agent message for repository '%s' (session: %s, agent: %s)",
    name,
    session_id or "new",
    agent or "default",
)

# SOURCE: agent_service.py:22-27
# Pattern for thread-safe state management
_processing_lock = Lock()
_processing_channels: dict[str, datetime] = {}

---

Edge Cases & Risks

Risk/Edge Case	Mitigation
sessionId missing from request	Add validation: return 400 Bad Request if sessionId is not provided
sessionId collision (same ID for different users)	Not a concern: if same user wants two sessions, they need different IDs; backend doesn't control ID generation
Channel name with colon in repo name	Use rsplit(":", 1) to split only on last colon; handles edge case of repo names containing :
Existing tests breaking	Update tests to include sessionId parameter; add new concurrent session tests
Backend state growth with abandoned sessions	Sessions should be cleaned up on disconnect/timeout (future enhancement)
History export needs repo name extraction	Already uses separate channel parameter; change applies to working directory only

---

Validation

Automated Checks

cd packages/pybackend && uv run pytest tests/unit/test_api.py -v
cd packages/pybackend && uv run pytest tests/unit/test_agent_service.py -v
cd packages/pybackend && uv run ruff check agent_service.py app.py
cd packages/pybackend && uv run mypy agent_service.py app.py

Manual Verification

1. Start backend with make run or cd packages/pybackend && uv run uvicorn app:app --host 0.0.0.0 --port 3000
2. Send two concurrent agent requests to same repository with different sessionIds
3. Verify both requests succeed (HTTP 200) instead of second getting 409 Conflict
4. Verify status endpoint works with sessionId parameter: GET /api/repositories/{name}/agent/status?sessionId={id}
5. Verify cancel endpoint works with sessionId parameter: POST /api/repositories/{name}/agent/cancel?sessionId={id}

---

Scope Boundaries

IN SCOPE:

• Modify channel identifiers to include session_id for all agent types
• Add sessionId as required query parameter to status/cancel endpoints
• Update _get_working_directory to extract repo name from composite channel
• Add helper function for channel parsing
• Update and add unit tests

OUT OF SCOPE (do not touch):

• Frontend changes (already sends sessionId in payload body)
• Session cleanup or timeout handling (future enhancement)
• Rate limiting or resource quotas
• Multi-session coordination within same session (multiple concurrent requests per session)
• History export functionality (already uses separate channel parameter)
• Agent CLI subprocess handling

---

Metadata

• Investigated by: GHAR
• Timestamp: 2026-04-03T00:00:00Z
• Artifact: .ghar/issues/issue-360.md