scripts-session-linker
#!/usr/bin/env python3 """​
title: "Session Linker - Cross-Session Context Linking" component_type: script version: "1.0.0" audience: contributor status: active summary: "J.8.3: Cross-session context linking with automatic continuation detection and chain traversal" keywords: ['session', 'linking', 'continuity', 'chain', 'context', 'cross-session'] tokens: ~400 created: 2026-02-04 updated: 2026-02-04 track: J task_id: J.8.3​
Session Linker - Cross-Session Context Linking (J.8.3)
Provides:
- Manual session linking (--link A B relationship)
- Session link queries (--links SESSION)
- Automatic continuation detection (--auto-link)
- Session chain traversal (--chain SESSION)
Relationships:
- continues: Session B continues work from session A
- references: Session B references session A
- supersedes: Session B replaces/supersedes session A
- fixes: Session B fixes issues from session A """
import sqlite3 from dataclasses import dataclass, field from datetime import datetime, timedelta from pathlib import Path from typing import List, Dict, Any, Optional, Tuple, Set import json import re
@dataclass class SessionLink: """Represents a link between two sessions.""" session_a: str session_b: str relationship: str notes: Optional[str] = None created_at: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
return {
'session_a': self.session_a,
'session_b': self.session_b,
'relationship': self.relationship,
'notes': self.notes,
'created_at': self.created_at
}
@dataclass class SessionInfo: """Summary information about a session.""" session_id: str message_count: int first_message: Optional[str] = None last_message: Optional[str] = None topics: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return {
'session_id': self.session_id,
'message_count': self.message_count,
'first_message': self.first_message,
'last_message': self.last_message,
'topics': self.topics
}
Valid relationship types
VALID_RELATIONSHIPS = {'continues', 'references', 'supersedes', 'fixes'}
class SessionLinker: """ Manages cross-session linking for context continuity.
Stores links in the session_links table (sessions.db, Tier 3).
"""
def __init__(self, sessions_db_path: Path):
"""Initialize with path to sessions.db."""
self.db_path = sessions_db_path
self._ensure_table()
def _ensure_table(self) -> None:
"""Ensure session_links table exists with proper schema."""
conn = sqlite3.connect(self.db_path)
try:
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS session_links (
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_a TEXT NOT NULL,
session_b TEXT NOT NULL,
relationship TEXT NOT NULL,
notes TEXT,
created_at TEXT DEFAULT CURRENT_TIMESTAMP,
UNIQUE(session_a, session_b, relationship)
)
""")
# Create indexes for efficient querying
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_session_links_a
ON session_links(session_a)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_session_links_b
ON session_links(session_b)
""")
conn.commit()
finally:
conn.close()
def create_link(
self,
session_a: str,
session_b: str,
relationship: str,
notes: Optional[str] = None
) -> Tuple[bool, str]:
"""
Create a link between two sessions.
Args:
session_a: Source session ID
session_b: Target session ID
relationship: Type of relationship (continues, references, supersedes, fixes)
notes: Optional notes about the link
Returns:
Tuple of (success, message)
"""
if relationship not in VALID_RELATIONSHIPS:
return False, f"Invalid relationship '{relationship}'. Valid: {', '.join(sorted(VALID_RELATIONSHIPS))}"
if session_a == session_b:
return False, "Cannot link a session to itself"
conn = sqlite3.connect(self.db_path)
try:
cursor = conn.cursor()
cursor.execute("""
INSERT INTO session_links (session_a, session_b, relationship, notes)
VALUES (?, ?, ?, ?)
""", (session_a, session_b, relationship, notes))
conn.commit()
return True, f"Created link: {session_a} --[{relationship}]--> {session_b}"
except sqlite3.IntegrityError:
return False, f"Link already exists: {session_a} --[{relationship}]--> {session_b}"
finally:
conn.close()
def get_links(self, session_id: str, direction: str = 'both') -> List[SessionLink]:
"""
Get all links for a session.
Args:
session_id: Session ID to query
direction: 'outgoing' (session_a), 'incoming' (session_b), or 'both'
Returns:
List of SessionLink objects
"""
conn = sqlite3.connect(self.db_path)
try:
cursor = conn.cursor()
links = []
if direction in ('outgoing', 'both'):
cursor.execute("""
SELECT session_a, session_b, relationship, notes, created_at
FROM session_links
WHERE session_a LIKE ?
ORDER BY created_at DESC
""", (f"%{session_id}%",))
for row in cursor.fetchall():
links.append(SessionLink(*row))
if direction in ('incoming', 'both'):
cursor.execute("""
SELECT session_a, session_b, relationship, notes, created_at
FROM session_links
WHERE session_b LIKE ?
ORDER BY created_at DESC
""", (f"%{session_id}%",))
for row in cursor.fetchall():
links.append(SessionLink(*row))
# Deduplicate
seen = set()
unique_links = []
for link in links:
key = (link.session_a, link.session_b, link.relationship)
if key not in seen:
seen.add(key)
unique_links.append(link)
return unique_links
finally:
conn.close()
def get_chain(
self,
session_id: str,
relationship: str = 'continues',
direction: str = 'backward',
max_depth: int = 10
) -> List[str]:
"""
Traverse session chain following a relationship type.
Args:
session_id: Starting session
relationship: Relationship to follow (default: continues)
direction: 'backward' (find predecessors) or 'forward' (find successors)
max_depth: Maximum chain length
Returns:
List of session IDs in chain order
"""
conn = sqlite3.connect(self.db_path)
try:
cursor = conn.cursor()
chain = [session_id]
visited = {session_id}
current = session_id
for _ in range(max_depth):
if direction == 'backward':
# Find session that this one continues from
cursor.execute("""
SELECT session_a FROM session_links
WHERE session_b LIKE ? AND relationship = ?
LIMIT 1
""", (f"%{current}%", relationship))
else:
# Find session that continues from this one
cursor.execute("""
SELECT session_b FROM session_links
WHERE session_a LIKE ? AND relationship = ?
LIMIT 1
""", (f"%{current}%", relationship))
row = cursor.fetchone()
if not row:
break
next_session = row[0]
if next_session in visited:
break # Cycle detected
visited.add(next_session)
if direction == 'backward':
chain.insert(0, next_session)
else:
chain.append(next_session)
current = next_session
return chain
finally:
conn.close()
def get_full_chain(self, session_id: str, relationship: str = 'continues') -> List[str]:
"""Get the complete chain in both directions."""
backward = self.get_chain(session_id, relationship, 'backward')
forward = self.get_chain(session_id, relationship, 'forward')
# Merge: backward already includes session_id, forward starts with it
return backward + forward[1:] # Skip duplicate session_id
def auto_detect_continuations(self, hours: int = 24) -> List[Tuple[str, str, float]]:
"""
Automatically detect session continuations based on:
- Temporal proximity (within hours)
- Topic similarity
- Explicit "continuation" phrases in messages
Args:
hours: Look for continuations within this many hours
Returns:
List of (session_a, session_b, confidence) tuples
"""
conn = sqlite3.connect(self.db_path)
try:
cursor = conn.cursor()
# Get sessions with their time ranges
cursor.execute("""
SELECT session_id,
MIN(timestamp) as first_msg,
MAX(timestamp) as last_msg,
COUNT(*) as msg_count
FROM messages
WHERE timestamp > datetime('now', ?)
GROUP BY session_id
ORDER BY first_msg
""", (f"-{hours * 2} hours",))
sessions = cursor.fetchall()
candidates = []
for i in range(len(sessions) - 1):
curr_id, curr_first, curr_last, curr_count = sessions[i]
next_id, next_first, next_last, next_count = sessions[i + 1]
# Skip if already linked
cursor.execute("""
SELECT 1 FROM session_links
WHERE session_a = ? AND session_b = ? AND relationship = 'continues'
""", (curr_id, next_id))
if cursor.fetchone():
continue
# Calculate confidence based on temporal proximity
try:
curr_last_dt = datetime.fromisoformat(curr_last.replace('Z', '+00:00'))
next_first_dt = datetime.fromisoformat(next_first.replace('Z', '+00:00'))
gap_hours = (next_first_dt - curr_last_dt).total_seconds() / 3600
if gap_hours < 0 or gap_hours > hours:
continue
# Confidence decreases with gap
temporal_confidence = max(0, 1 - (gap_hours / hours))
except (ValueError, TypeError):
continue
# Check for continuation phrases in first message of next session
cursor.execute("""
SELECT content FROM messages
WHERE session_id = ?
ORDER BY timestamp ASC
LIMIT 1
""", (next_id,))
first_msg = cursor.fetchone()
phrase_confidence = 0.0
if first_msg and first_msg[0]:
content = first_msg[0].lower()
continuation_phrases = [
'continuing from', 'continuing where', 'picking up where',
'as discussed', 'following up', 'resuming',
'back to the', 'returning to', 'let\'s continue',
'session continuation', 'context from previous'
]
if any(phrase in content for phrase in continuation_phrases):
phrase_confidence = 0.4
# Combined confidence
confidence = min(1.0, temporal_confidence * 0.6 + phrase_confidence + 0.2)
if confidence >= 0.3: # Minimum threshold
candidates.append((curr_id, next_id, round(confidence, 2)))
return candidates
finally:
conn.close()
def auto_link(self, min_confidence: float = 0.5, hours: int = 24, dry_run: bool = False) -> List[Dict[str, Any]]:
"""
Automatically create continuation links based on detection.
Args:
min_confidence: Minimum confidence threshold (0-1)
hours: Time window to search
dry_run: If True, don't actually create links
Returns:
List of created/proposed links with metadata
"""
candidates = self.auto_detect_continuations(hours)
results = []
for session_a, session_b, confidence in candidates:
if confidence < min_confidence:
continue
result = {
'session_a': session_a,
'session_b': session_b,
'relationship': 'continues',
'confidence': confidence,
'created': False
}
if not dry_run:
success, msg = self.create_link(
session_a,
session_b,
'continues',
notes=f"Auto-detected (confidence: {confidence})"
)
result['created'] = success
result['message'] = msg
else:
result['message'] = f"Would create: {session_a} --[continues]--> {session_b}"
results.append(result)
return results
def get_session_info(self, session_id: str) -> Optional[SessionInfo]:
"""Get summary information about a session."""
conn = sqlite3.connect(self.db_path)
try:
cursor = conn.cursor()
# Get message count and time range
cursor.execute("""
SELECT COUNT(*), MIN(timestamp), MAX(timestamp)
FROM messages
WHERE session_id LIKE ?
""", (f"%{session_id}%",))
row = cursor.fetchone()
if not row or row[0] == 0:
return None
count, first, last = row
# Extract topics from first few messages
cursor.execute("""
SELECT content FROM messages
WHERE session_id LIKE ? AND role = 'user'
ORDER BY timestamp
LIMIT 5
""", (f"%{session_id}%",))
topics = []
for (content,) in cursor.fetchall():
if content:
# Simple topic extraction: first 50 chars
topic = content[:50].replace('\n', ' ').strip()
if topic:
topics.append(topic + ('...' if len(content) > 50 else ''))
return SessionInfo(
session_id=session_id,
message_count=count,
first_message=first,
last_message=last,
topics=topics[:3] # Top 3 topics
)
finally:
conn.close()
def get_link_stats(self) -> Dict[str, Any]:
"""Get statistics about session links."""
conn = sqlite3.connect(self.db_path)
try:
cursor = conn.cursor()
cursor.execute("""
SELECT relationship, COUNT(*) as count
FROM session_links
GROUP BY relationship
ORDER BY count DESC
""")
by_relationship = {row[0]: row[1] for row in cursor.fetchall()}
cursor.execute("SELECT COUNT(*) FROM session_links")
total = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(DISTINCT session_a) FROM session_links")
unique_sources = cursor.fetchone()[0]
cursor.execute("SELECT COUNT(DISTINCT session_b) FROM session_links")
unique_targets = cursor.fetchone()[0]
return {
'total_links': total,
'by_relationship': by_relationship,
'unique_source_sessions': unique_sources,
'unique_target_sessions': unique_targets
}
finally:
conn.close()
def format_links_output(links: List[SessionLink], json_output: bool = False) -> str: """Format session links for display.""" if json_output: return json.dumps([link.to_dict() for link in links], indent=2)
if not links:
return "No session links found."
lines = ["Session Links:", "=" * 60]
for link in links:
arrow = "--[" + link.relationship + "]-->"
lines.append(f"{link.session_a[:12]}... {arrow} {link.session_b[:12]}...")
if link.notes:
lines.append(f" Notes: {link.notes}")
if link.created_at:
lines.append(f" Created: {link.created_at}")
lines.append("")
return "\n".join(lines)
def format_chain_output(chain: List[str], linker: SessionLinker, json_output: bool = False) -> str: """Format session chain for display.""" if json_output: return json.dumps({ 'chain': chain, 'length': len(chain), 'sessions': [ linker.get_session_info(s).to_dict() if linker.get_session_info(s) else {'session_id': s} for s in chain ] }, indent=2)
if not chain:
return "No session chain found."
lines = [f"Session Chain ({len(chain)} sessions):", "=" * 60]
for i, session_id in enumerate(chain):
info = linker.get_session_info(session_id)
prefix = " " if i > 0 else ""
arrow = " --> " if i < len(chain) - 1 else ""
if info:
lines.append(f"{prefix}{session_id[:16]}... ({info.message_count} msgs)")
if info.topics:
lines.append(f"{prefix} Topics: {info.topics[0]}")
else:
lines.append(f"{prefix}{session_id[:16]}... (no data)")
if arrow:
lines.append(f"{prefix} |")
lines.append(f"{prefix} v")
return "\n".join(lines)
def format_link_help() -> str: """Return help text for session linking.""" return """ Session Linking (J.8.3) - Cross-Session Context Continuity
COMMANDS: --link A B REL Create a link between sessions A and B --links SESSION Show all links for a session --chain SESSION Show the full session chain --auto-link Auto-detect and create continuation links --link-stats Show session link statistics
RELATIONSHIPS: continues Session B continues work from session A references Session B references content from session A supersedes Session B replaces/supersedes session A fixes Session B fixes issues found in session A
EXAMPLES:
Create a continuation link
/cxq --link abc123 def456 continues
Show all sessions linked to a session
/cxq --links abc123
Show full session chain (all continuations)
/cxq --chain abc123
Auto-detect continuations (dry run)
/cxq --auto-link --dry-run
Auto-detect with custom threshold
/cxq --auto-link --min-confidence 0.6
AUTO-DETECTION: The auto-link feature detects continuations based on:
- Temporal proximity (sessions within 24 hours)
- Continuation phrases in messages
- Gap time between sessions
Confidence thresholds:
- 0.7+ High confidence (likely continuation)
- 0.5-0.7 Medium confidence (probable)
- 0.3-0.5 Low confidence (possible) """
CLI entry point for standalone testing
if name == "main": import sys
# Find sessions.db
try:
from scripts.core.paths import get_sessions_db_path
db_path = get_sessions_db_path()
except ImportError:
db_path = Path.home() / "PROJECTS" / ".coditect-data" / "context-storage" / "sessions.db"
linker = SessionLinker(db_path)
if len(sys.argv) < 2:
print(format_link_help())
sys.exit(0)
cmd = sys.argv[1]
if cmd == "--stats":
stats = linker.get_link_stats()
print(json.dumps(stats, indent=2))
elif cmd == "--links" and len(sys.argv) > 2:
links = linker.get_links(sys.argv[2])
print(format_links_output(links))
elif cmd == "--chain" and len(sys.argv) > 2:
chain = linker.get_full_chain(sys.argv[2])
print(format_chain_output(chain, linker))
elif cmd == "--auto-link":
dry_run = "--dry-run" in sys.argv
results = linker.auto_link(dry_run=dry_run)
print(json.dumps(results, indent=2))
elif cmd == "--link" and len(sys.argv) >= 5:
session_a, session_b, rel = sys.argv[2], sys.argv[3], sys.argv[4]
success, msg = linker.create_link(session_a, session_b, rel)
print(msg)
else:
print(format_link_help())