-
Notifications
You must be signed in to change notification settings - Fork 0
Description
Problem
The Session Coordinator's primitive operations (store_data, retrieve_data) are too verbose for common workflows. Simple operations like "mark batch complete" require multiple tool calls with manual state manipulation:
Current approach (4+ tool calls):
1. retrieve_data("issue:54", "current_batch") → 2
2. retrieve_data("issue:54", "todos") → [...]
3. Manually modify data in memory
4. store_data("issue:54", "current_batch", 3)
5. store_data("issue:54", "todos", [...])
This creates excessive overhead and makes the Session Coordinator tedious to use during active development.
Solution
Add a convenience operations layer on top of primitives:
- High-level operations for common patterns
- Single tool call handles retrieve + modify + store
- Primitives still available for custom needs
- Reduces verbosity by 75%+ for typical workflows
Architecture
Three-Layer Design
Layer 3: Convenience Operations (High-Level)
↓
Layer 2: Primitives (Low-Level)
↓
Layer 1: Storage Adapter
Layer 1 (Storage): File system, Redis, etc.
Layer 2 (Primitives): store_data, retrieve_data, etc.
Layer 3 (Convenience): complete_batch, add_commit, etc.
Users can choose their level of abstraction.
Proposed Operations
Priority: High (Must Have)
1. complete_batch(issue, batch)
Mark batch complete, advance to next batch.
Usage:
complete_batch(54, 2)
# Marks batch 2 todos complete
# Sets current_batch = 3
# Returns: {batch_completed: 2, next_batch: 3}Replaces:
# Get current state
todos = retrieve_data("issue:54", "todos")
current_batch = retrieve_data("issue:54", "current_batch")
# Modify
for todo in todos:
if todo["batch"] == 2:
todo["complete"] = True
# Store
store_data("issue:54", "todos", todos)
store_data("issue:54", "current_batch", 3)Implementation:
def complete_batch(issue_number: int, batch_number: int) -> Dict[str, Any]:
"""
Mark all todos in a batch as complete and advance to next batch.
Args:
issue_number: GitHub issue number
batch_number: Batch to mark complete
Returns:
{
"batch_completed": int,
"next_batch": int,
"todos_updated": int
}
"""
scope = f"issue:{issue_number}"
# Retrieve
todos = adapter.retrieve(scope, "todos") or []
# Modify
updated_count = 0
for todo in todos:
if todo.get("batch") == batch_number:
todo["complete"] = True
updated_count += 1
next_batch = batch_number + 1
# Store
adapter.store(scope, "todos", todos)
adapter.store(scope, "current_batch", next_batch)
return {
"batch_completed": batch_number,
"next_batch": next_batch,
"todos_updated": updated_count
}2. add_commit(issue, commit_hash, batch?)
Append commit to issue's commit history.
Usage:
add_commit(54, "abc123def", batch=2)
# Appends to commits array
# Optionally associates with batch
# Returns: {commits_total: 5}Replaces:
commits = retrieve_data("issue:54", "commits") or []
commits.append({"hash": "abc123def", "batch": 2})
store_data("issue:54", "commits", commits)Implementation:
def add_commit(issue_number: int, commit_hash: str, batch: int = None) -> Dict[str, Any]:
"""
Add a commit hash to the issue's commit history.
Args:
issue_number: GitHub issue number
commit_hash: Git commit hash
batch: Optional batch number this commit belongs to
Returns:
{"commits_total": int}
"""
scope = f"issue:{issue_number}"
# Retrieve
commits = adapter.retrieve(scope, "commits") or []
# Build commit record
commit_record = {"hash": commit_hash}
if batch is not None:
commit_record["batch"] = batch
# Append and store
commits.append(commit_record)
adapter.store(scope, "commits", commits)
return {"commits_total": len(commits)}3. update_status(issue, status, notes?)
Update issue status with optional notes.
Usage:
update_status(54, "paused", "Waiting for design feedback")
# Returns: {status: "paused"}Replaces:
store_data("issue:54", "status", "paused")
store_data("issue:54", "status_notes", "Waiting for design feedback")Implementation:
def update_status(issue_number: int, status: str, notes: str = None) -> Dict[str, Any]:
"""
Update issue status.
Args:
issue_number: GitHub issue number
status: One of "in_progress", "paused", "blocked", "complete"
notes: Optional status notes
Returns:
{"status": str}
"""
scope = f"issue:{issue_number}"
# Validate status
valid_statuses = ["in_progress", "paused", "blocked", "complete"]
if status not in valid_statuses:
raise ValueError(f"Status must be one of: {valid_statuses}")
# Store
adapter.store(scope, "status", status)
if notes:
adapter.store(scope, "status_notes", notes)
return {"status": status}Priority: Medium (Should Have)
4. load_issue_state(issue) / save_issue_state(issue, state)
Load all issue state at once for intensive work.
Usage:
# At start of work session
state = load_issue_state(54)
# Returns: {status: "in_progress", current_batch: 2, todos: [...], ...}
# Work in memory (no tool calls)
state["current_batch"] = 3
state["todos"][5]["complete"] = True
state["commits"].append("abc123")
# At end or milestone
save_issue_state(54, state)Why this matters:
- Reduces tool calls from dozens to 2 (load + save)
- Better for intensive batch processing
- All modifications in memory (fast)
Implementation:
def load_issue_state(issue_number: int) -> Dict[str, Any]:
"""
Load all state for an issue into memory.
Returns complete state object that can be modified in memory
and saved back with save_issue_state().
"""
scope = f"issue:{issue_number}"
keys = adapter.list_keys(scope)
state = {}
for key in keys:
state[key] = adapter.retrieve(scope, key)
return state
def save_issue_state(issue_number: int, state: Dict[str, Any]) -> Dict[str, Any]:
"""
Save complete issue state.
Writes all state keys from the state object back to storage.
"""
scope = f"issue:{issue_number}"
for key, value in state.items():
adapter.store(scope, key, value)
return {"keys_saved": len(state)}5. append_to_array(scope, key, value)
Atomic append operation (doesn't need full retrieve).
Usage:
append_to_array("issue:54", "next_steps", "Review batch 3 PRs")
append_to_array("issue:54", "blockers", "Need API key from ops")Implementation:
def append_to_array(scope: str, key: str, value: Any) -> Dict[str, Any]:
"""
Append a value to an array without retrieving entire array.
Thread-safe atomic operation.
"""
current = adapter.retrieve(scope, key) or []
current.append(value)
adapter.store(scope, key, current)
return {"array_length": len(current)}6. increment(scope, key, amount=1)
Atomic increment operation.
Usage:
increment("issue:54", "current_batch") # +1
increment("session:cipher", "batches_completed", 1)Implementation:
def increment(scope: str, key: str, amount: int = 1) -> Dict[str, Any]:
"""
Increment a numeric value atomically.
"""
current = adapter.retrieve(scope, key) or 0
new_value = current + amount
adapter.store(scope, key, new_value)
return {"new_value": new_value}Priority: Low (Nice to Have)
7. bulk_update(scope, updates)
Update multiple keys in single call.
Usage:
bulk_update("issue:54", {
"status": "in_progress",
"current_batch": 3,
"worktree": ".worktrees/issue-54"
})Implementation Plan
Phase 1: Core Convenience Operations (2-3 hours)
- Create
tools/convenience.py - Implement
complete_batch() - Implement
add_commit() - Implement
update_status() - Register tools in MCP server
- Write tests
Phase 2: State Management Operations (1-2 hours)
- Implement
load_issue_state() - Implement
save_issue_state() - Write tests for state round-tripping
- Document usage patterns
Phase 3: Atomic Operations (1 hour)
- Implement
append_to_array() - Implement
increment() - Write tests
- Document thread-safety considerations
Phase 4: Documentation & Examples (1 hour)
- Update README with convenience operations
- Add usage examples
- Document when to use primitives vs convenience
- Update
.claude/CLAUDE.mdtemplate
Total Estimated Time: 5-7 hours
File Structure
src/claude_session_coordinator/
├── tools/
│ ├── session.py # sign_on, sign_off
│ ├── data.py # Primitives: store, retrieve, delete
│ ├── discovery.py # list_keys, list_scopes
│ └── convenience.py # NEW: High-level operations
Usage Comparison
Before (Primitives Only)
# Claude Code making 6+ tool calls to advance batch
retrieve_data("issue:54", "current_batch") # → 2
retrieve_data("issue:54", "todos") # → [...]
retrieve_data("issue:54", "commits") # → [...]
# Manual modification in Claude's mind
# Mark batch 2 todos complete
# Prepare updated todos array
store_data("issue:54", "current_batch", 3)
store_data("issue:54", "todos", updated_todos)
store_data("issue:54", "commits", [..."abc123"])
# 6 tool calls minimumAfter (Convenience Operations)
# Single operation
complete_batch(54, 2)
# Separate commit tracking
add_commit(54, "abc123", batch=2)
# 2 tool calls totalReduction: 70% fewer tool calls!
Backwards Compatibility
- ✅ Primitives remain available
- ✅ No breaking changes
- ✅ Convenience operations built on top of primitives
- ✅ Users can mix and match as needed
Testing Strategy
Unit Tests
def test_complete_batch():
"""Test batch completion updates state correctly."""
# Setup: Create issue with todos
# Execute: complete_batch(54, 2)
# Assert: todos marked complete, current_batch = 3
def test_add_commit():
"""Test commit appending."""
# Execute: add_commit(54, "abc123", batch=2)
# Assert: commit in array with correct batch
def test_load_save_state_roundtrip():
"""Test state loads and saves correctly."""
# Load state, modify, save, load again
# Assert: modifications persistedIntegration Tests
def test_typical_workflow():
"""Test typical issue workflow with convenience ops."""
# sign_on()
# load_issue_state()
# modify in memory
# save_issue_state()
# complete_batch()
# add_commit()
# update_status("complete")
# sign_off()
# Assert: All operations work together correctlyDocumentation Updates
README.md Section
## Convenience Operations
The Session Coordinator provides high-level operations for common patterns:
### Complete a Batch
```python
complete_batch(54, 2)
# Marks batch 2 complete, advances to batch 3
```
### Track Commits
```python
add_commit(54, "abc123", batch=2)
# Adds commit to issue history
```
### Update Status
```python
update_status(54, "paused", "Waiting for review")
```
### Work with State
```python
# Load state once
state = load_issue_state(54)
# Modify in memory (fast)
state["current_batch"] = 3
state["todos"][5]["complete"] = True
# Save once
save_issue_state(54, state)
```
**When to use:**
- Convenience operations: Common workflows (80% of use cases)
- State management: Intensive batch processing
- Primitives: Custom needs, special patternsSuccess Criteria
- Typical workflows reduced from 6+ tool calls to 1-2
- Convenience operations tested and documented
- Primitives still available for custom needs
- No breaking changes to existing API
- Clear documentation on when to use each layer
Related Issues
- Issue Initial Implementation: Core MCP Server for Session Coordination #1: Core MCP Server (provides primitives)
- Issue docs(workflow): add standardized 'Resume Issue Workflow' template #58: Workflow integration (uses these operations)
- Issue #23: Dev Squad registry (coordination patterns)
Why This Matters
The Session Coordinator is only useful if it's easy to use during active development. The primitive operations are correct but too verbose for practical workflows.
Convenience operations make coordination feel natural instead of tedious.
Before: "Ugh, I need to make 6 tool calls to mark a batch complete"
After: "Just call complete_batch() and keep working"
This transforms the Session Coordinator from "technically correct" to "actually useful."
Priority: High
Estimated Effort: 5-7 hours
Dependencies: Issue #1 (primitives must exist first)
Assignee: Cipher (DS-02) - Technical implementation specialist
"According to my calculations, these convenience operations will reduce tool call overhead by 75% in typical workflows." - Cipher (DS-02)