---
phase: 05-retrieval-orchestrator
plan: "02"
type: execute
wave: 2
depends_on: ["05-01"]
files_modified:
- src/skill_retriever/workflows/pipeline.py
- src/skill_retriever/workflows/dependency_resolver.py
- src/skill_retriever/workflows/__init__.py
- tests/test_dependency_resolver.py
- tests/test_pipeline.py
autonomous: true
must_haves:
truths:
- "Given recommended components, all transitive dependencies are included in result"
- "Conflicts between recommended components are detected and surfaced"
- "Dependency resolution handles cycles gracefully without infinite loops"
- "Missing graph nodes in dependency chain are handled without crashing"
artifacts:
- path: "src/skill_retriever/workflows/dependency_resolver.py"
provides: "Transitive dependency resolution and conflict detection"
exports: ["resolve_transitive_dependencies", "detect_conflicts"]
- path: "tests/test_dependency_resolver.py"
provides: "Tests for dependency resolution and conflict detection"
min_lines: 60
key_links:
- from: "src/skill_retriever/workflows/dependency_resolver.py"
to: "networkx.descendants"
via: "nx.descendants() for transitive closure"
pattern: "nx\\.descendants"
- from: "src/skill_retriever/workflows/dependency_resolver.py"
to: "EdgeType.CONFLICTS_WITH"
via: "Edge type check for conflict detection"
pattern: "EdgeType\\.CONFLICTS_WITH"
- from: "src/skill_retriever/workflows/pipeline.py"
to: "dependency_resolver"
via: "Pipeline calls resolve and detect before context assembly"
pattern: "resolve_transitive_dependencies|detect_conflicts"
---
<objective>
Add transitive dependency resolution and conflict detection to complete the orchestration layer, fulfilling requirements GRPH-02, GRPH-03, and GRPH-04.
Purpose: Ensures returned component sets are complete (no missing pieces) and conflict-free (issues surfaced before installation).
Output: Working dependency resolver that uses nx.descendants() for transitive closure, conflict detector that finds CONFLICTS_WITH edges, and integration with RetrievalPipeline.
</objective>
<execution_context>
@C:\Users\33641\.claude/get-shit-done/workflows/execute-plan.md
@C:\Users\33641\.claude/get-shit-done/templates/summary.md
</execution_context>
<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/05-retrieval-orchestrator/05-RESEARCH.md
# Plan 01 artifacts (must be complete first)
@src/skill_retriever/workflows/pipeline.py
@src/skill_retriever/workflows/models.py
# Memory layer for graph access
@src/skill_retriever/memory/graph_store.py
# Entities for EdgeType
@src/skill_retriever/entities/graph.py
</context>
<tasks>
<task type="auto">
<name>Task 1: Create dependency resolver with transitive closure and conflict detection</name>
<files>src/skill_retriever/workflows/dependency_resolver.py, src/skill_retriever/workflows/__init__.py, tests/test_dependency_resolver.py</files>
<action>
Create `src/skill_retriever/workflows/dependency_resolver.py` with two functions:
**1. `resolve_transitive_dependencies()` function:**
```python
def resolve_transitive_dependencies(
component_ids: list[str],
graph_store: NetworkXGraphStore,
) -> tuple[set[str], list[str]]:
"""Resolve all transitive dependencies for the given components.
Returns:
Tuple of (all_component_ids, newly_added_dependency_ids)
- all_component_ids: Original components plus all their transitive deps
- newly_added_dependency_ids: Only the deps that weren't in original list
"""
```
Implementation:
- Use `isinstance(graph_store, NetworkXGraphStore)` to access `._graph` attribute
- For each component_id, check if node exists in graph first (pitfall #5 from research)
- Use `nx.descendants(graph._graph, node_id)` for transitive closure
- Filter descendants to only include nodes reachable via DEPENDS_ON edges:
- Build a subgraph of only DEPENDS_ON edges
- Run descendants on that subgraph
- Handle cycles: Use `nx.is_directed_acyclic_graph()` check, log warning if cycle detected
- Return tuple of (all_ids including originals, only the newly added deps)
**2. `detect_conflicts()` function:**
```python
def detect_conflicts(
component_ids: set[str],
graph_store: NetworkXGraphStore,
) -> list[ConflictInfo]:
"""Find all CONFLICTS_WITH relationships among the given components.
Checks both directions (A conflicts B and B conflicts A are the same conflict).
"""
```
Implementation:
- For each component_id, call `graph_store.get_edges(component_id)`
- Check edges where `edge.edge_type == EdgeType.CONFLICTS_WITH`
- If target/source is in component_ids set, record conflict
- Use `frozenset({a, b})` in a `checked` set to avoid duplicate pairs (research pattern)
- Extract reason from edge.metadata.get("reason", "Component conflict detected")
Create `tests/test_dependency_resolver.py` with tests:
- `test_resolve_empty_list` - Empty input returns empty output
- `test_resolve_no_deps` - Component with no DEPENDS_ON edges returns only itself
- `test_resolve_single_hop` - A depends on B, returns {A, B}
- `test_resolve_transitive` - A depends on B, B depends on C, returns {A, B, C}
- `test_resolve_missing_node` - Component not in graph is handled gracefully
- `test_detect_no_conflicts` - Components with no conflicts returns empty list
- `test_detect_single_conflict` - A conflicts with B, both in set, returns ConflictInfo
- `test_detect_bidirectional_check` - Conflict stored A->B still found when checking B
- `test_detect_excludes_non_selected` - A conflicts with B, but B not in selected set, no conflict
Update `__init__.py` to export:
- `resolve_transitive_dependencies`
- `detect_conflicts`
</action>
<verify>
```bash
cd C:/Users/33641/repos/skill-retriever
uv run pytest tests/test_dependency_resolver.py -v
uv run ruff check src/skill_retriever/workflows/
uv run pyright src/skill_retriever/workflows/
```
</verify>
<done>resolve_transitive_dependencies and detect_conflicts pass all tests, handle edge cases</done>
</task>
<task type="auto">
<name>Task 2: Integrate dependency resolution and conflict detection into pipeline</name>
<files>src/skill_retriever/workflows/pipeline.py, tests/test_pipeline.py</files>
<action>
Update `src/skill_retriever/workflows/pipeline.py` to integrate dependency resolution:
1. **Add new stages to retrieve() method:**
- After score fusion and BEFORE context assembly:
- Stage 5a: Call `resolve_transitive_dependencies()` on fused component IDs
- Stage 5b: Call `detect_conflicts()` on the expanded component set
- Stage 6: Call `assemble_context()` with the expanded component list
2. **Update cache key:**
- Cache invalidation needed if dependencies change
- For now, cache at query level is fine (deps are deterministic from graph)
3. **Populate PipelineResult fields:**
- `dependencies_added`: List of component IDs that were added via transitive resolution
- `conflicts`: List of ConflictInfo from detect_conflicts()
4. **Important: Resolve deps BEFORE budget check (research pitfall #2)**
- Pass expanded component list to assemble_context()
- Let assembler handle truncation within budget
5. **Handle NetworkXGraphStore type narrowing:**
- Use isinstance check for accessing ._graph attribute
- If not NetworkXGraphStore, skip dependency resolution (return original list)
Update `tests/test_pipeline.py` with additional tests:
- `test_pipeline_resolves_dependencies` - Component with deps gets deps included
- `test_pipeline_dependencies_added_populated` - dependencies_added list is accurate
- `test_pipeline_detects_conflicts` - Components with conflicts get ConflictInfo in result
- `test_pipeline_latency_under_1000ms` - Complex query with deps < 1000ms
</action>
<verify>
```bash
cd C:/Users/33641/repos/skill-retriever
uv run pytest tests/test_pipeline.py -v
uv run pytest -v # All tests
uv run ruff check src/
uv run pyright src/
```
</verify>
<done>Pipeline integrates dependency resolution, all tests pass, latency SLAs verified</done>
</task>
</tasks>
<verification>
```bash
cd C:/Users/33641/repos/skill-retriever
# All tests pass
uv run pytest -v
# Lint clean
uv run ruff check src/
# Types clean
uv run pyright src/
# Verify latency SLA (should see latency_ms in output)
uv run python -c "
from skill_retriever.memory.graph_store import NetworkXGraphStore
from skill_retriever.memory.vector_store import FAISSVectorStore
from skill_retriever.workflows import RetrievalPipeline
gs = NetworkXGraphStore()
vs = FAISSVectorStore(dimension=384)
pipeline = RetrievalPipeline(gs, vs)
result = pipeline.retrieve('test query')
print(f'Latency: {result.latency_ms:.1f}ms')
assert result.latency_ms < 1000, 'Latency SLA violated'
"
```
</verification>
<success_criteria>
- Transitive dependencies resolved via nx.descendants() on DEPENDS_ON subgraph
- Conflicts detected via CONFLICTS_WITH edge traversal (both directions)
- Pipeline returns complete component sets with dependencies_added populated
- Conflicts surfaced in PipelineResult.conflicts before user sees results
- Simple queries < 500ms, complex queries < 1000ms
- All tests pass, lint clean, types clean
</success_criteria>
<output>
After completion, create `.planning/phases/05-retrieval-orchestrator/05-02-SUMMARY.md`
</output>
