add tmux team observer + member lifecycle events (T-139, D-75)
test / unit + widget + golden + a11y (push) Failing after 27s
test / integration_test (xvfb) (push) Has been skipped
test / bundle smoke (xvfb 5s) (push) Has been skipped
test / daemon subprocess + web WASM smoke (push) Has been skipped
test / dart doc (lib API) (push) Failing after 24s
test / unit + widget + golden + a11y (push) Failing after 27s
test / integration_test (xvfb) (push) Has been skipped
test / bundle smoke (xvfb 5s) (push) Has been skipped
test / daemon subprocess + web WASM smoke (push) Has been skipped
test / dart doc (lib API) (push) Failing after 24s
team_observer.dart is the single drift-containment point for Claude Code's experimental tmux team mode. It discovers the active team for a workspace (~/.claude/teams/<team>/config.json, matched by member cwd), polls `tmux -L clide list-panes -a`, and correlates live panes with the config's tmuxPaneId to emit TeamMemberBorn / TeamMemberDied — identity (name, agentType, model, colour, pane) comes from the config, so it's reliable regardless of transcript drift. Each teammate's subagent transcript is resolved best-effort and streamed on a per-agent MessageBus channel via TranscriptPublisher (TranscriptReader gains an explicit `file:` for this). The config<->transcript join is the fragile part: no shared key, so it uses a sibling .meta.json agentType when present, else zips members-by-joinedAt against files-by-mtime. This join needs validation against a live team run. App wiring + visible surfacing land with the teammate tiles (T-140). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,316 @@
|
||||
/// tmux agent-team observer (epic T-132, T-139, D-75).
|
||||
///
|
||||
/// THE single drift-containment point for Claude Code's experimental tmux
|
||||
/// team mode. Everything that reads CC's undocumented team artifacts lives
|
||||
/// here so a CC change only breaks one file.
|
||||
///
|
||||
/// # What's reliable vs. fragile
|
||||
/// - **Reliable — lifecycle + identity (config-driven).** A team writes
|
||||
/// `~/.claude/teams/<team>/config.json` listing each member with its
|
||||
/// `tmuxPaneId` (`%N`, empty for the lead), `name`, `agentType`, `model`,
|
||||
/// `color`, `cwd`, `joinedAt`. Polling `tmux -L clide list-panes -a` and
|
||||
/// correlating live pane ids with `tmuxPaneId` gives a dependable
|
||||
/// born/died signal and full identity — no transcript needed.
|
||||
/// - **Fragile — per-teammate transcript join.** A teammate's transcript is
|
||||
/// a subagent file `<munged-cwd>/<leadSessionId>/subagents/agent-<hex>.jsonl`
|
||||
/// whose only ids are a random hex (the filename) and a `slug`; it carries
|
||||
/// no `name`/`agentType`. The config's `agentId` is `<name>@<team>` — a
|
||||
/// different namespace — so there is no shared key. We join via a sibling
|
||||
/// `agent-<hex>.meta.json` (`{agentType}`) when present, else fall back to
|
||||
/// zipping members-by-`joinedAt` against files-by-mtime. This is the part
|
||||
/// most likely to drift; it needs validation against a live team run.
|
||||
library;
|
||||
|
||||
import 'dart:async';
|
||||
import 'dart:convert';
|
||||
import 'dart:io';
|
||||
|
||||
import 'package:clide/builtin/claude/src/transcript_publisher.dart';
|
||||
import 'package:clide/builtin/claude/src/transcript_reader.dart';
|
||||
import 'package:clide/kernel/src/events/bus.dart';
|
||||
import 'package:clide/kernel/src/events/message_bus.dart';
|
||||
import 'package:clide/kernel/src/events/types.dart';
|
||||
|
||||
/// One member of a team config.
|
||||
class TeamMember {
|
||||
const TeamMember({
|
||||
required this.agentId,
|
||||
required this.name,
|
||||
required this.agentType,
|
||||
required this.tmuxPaneId,
|
||||
this.model,
|
||||
this.color,
|
||||
this.cwd,
|
||||
this.joinedAt,
|
||||
});
|
||||
|
||||
/// Config agent id, `<name>@<team>`.
|
||||
final String agentId;
|
||||
final String name;
|
||||
final String agentType;
|
||||
|
||||
/// tmux pane id (`%N`); empty for the lead.
|
||||
final String tmuxPaneId;
|
||||
final String? model;
|
||||
final String? color;
|
||||
final String? cwd;
|
||||
final int? joinedAt;
|
||||
|
||||
bool get isLead => tmuxPaneId.isEmpty || agentType == 'team-lead';
|
||||
}
|
||||
|
||||
/// Parsed `~/.claude/teams/<team>/config.json`.
|
||||
class TeamConfig {
|
||||
const TeamConfig({
|
||||
required this.team,
|
||||
required this.leadSessionId,
|
||||
required this.members,
|
||||
required this.createdAt,
|
||||
});
|
||||
|
||||
final String team;
|
||||
final String leadSessionId;
|
||||
final List<TeamMember> members;
|
||||
final int createdAt;
|
||||
|
||||
/// Non-lead members (the panes we surface).
|
||||
List<TeamMember> get teammates => members.where((m) => !m.isLead).toList();
|
||||
|
||||
/// Parse a config; returns null on malformed JSON.
|
||||
static TeamConfig? parse(String teamDirName, String jsonStr) {
|
||||
Map<String, dynamic> d;
|
||||
try {
|
||||
d = jsonDecode(jsonStr) as Map<String, dynamic>;
|
||||
} catch (_) {
|
||||
return null;
|
||||
}
|
||||
final members = <TeamMember>[];
|
||||
for (final m in (d['members'] as List? ?? const [])) {
|
||||
if (m is! Map) continue;
|
||||
members.add(TeamMember(
|
||||
agentId: m['agentId'] as String? ?? '',
|
||||
name: m['name'] as String? ?? '',
|
||||
agentType: m['agentType'] as String? ?? '',
|
||||
tmuxPaneId: m['tmuxPaneId'] as String? ?? '',
|
||||
model: m['model'] as String?,
|
||||
color: m['color'] as String?,
|
||||
cwd: m['cwd'] as String?,
|
||||
joinedAt: (m['joinedAt'] as num?)?.toInt(),
|
||||
));
|
||||
}
|
||||
return TeamConfig(
|
||||
team: (d['name'] as String?) ?? teamDirName,
|
||||
leadSessionId: d['leadSessionId'] as String? ?? '',
|
||||
members: members,
|
||||
createdAt: (d['createdAt'] as num?)?.toInt() ?? 0,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Discover the active team config for [workspacePath]: the team (under
|
||||
/// [teamsBase]) any of whose members runs in [workspacePath], newest by
|
||||
/// `createdAt` when several match. Null if none.
|
||||
Future<TeamConfig?> discoverTeam(String workspacePath, {required String teamsBase}) async {
|
||||
final dir = Directory(teamsBase);
|
||||
if (!await dir.exists()) return null;
|
||||
TeamConfig? best;
|
||||
await for (final entity in dir.list()) {
|
||||
if (entity is! Directory) continue;
|
||||
final cfgFile = File('${entity.path}/config.json');
|
||||
if (!await cfgFile.exists()) continue;
|
||||
final cfg = TeamConfig.parse(entity.path.split('/').last, await cfgFile.readAsString());
|
||||
if (cfg == null) continue;
|
||||
if (!cfg.members.any((m) => m.cwd == workspacePath)) continue;
|
||||
if (best == null || cfg.createdAt > best.createdAt) best = cfg;
|
||||
}
|
||||
return best;
|
||||
}
|
||||
|
||||
/// Returns the set of live tmux pane ids on the `clide` socket. Injectable
|
||||
/// so tests don't shell out.
|
||||
typedef PaneLister = Future<Set<String>> Function();
|
||||
|
||||
class _LiveMember {
|
||||
_LiveMember(this.member, this.team, this.publisher);
|
||||
final TeamMember member;
|
||||
final String team;
|
||||
final TranscriptPublisher? publisher;
|
||||
}
|
||||
|
||||
/// Watches a workspace's tmux team and emits [TeamMemberBorn] /
|
||||
/// [TeamMemberDied] as panes appear/disappear, publishing each teammate's
|
||||
/// transcript onto the [MessageBus] under its per-agent channel.
|
||||
class TeamObserver {
|
||||
TeamObserver({
|
||||
required this.workspacePath,
|
||||
required DaemonBus events,
|
||||
required MessageBus messages,
|
||||
String? teamsBase,
|
||||
String? projectsBase,
|
||||
PaneLister? paneLister,
|
||||
Duration pollInterval = const Duration(seconds: 2),
|
||||
}) : _events = events,
|
||||
_messages = messages,
|
||||
_teamsBase = teamsBase ?? _defaultTeamsBase(),
|
||||
_projectsBase = projectsBase ?? _defaultProjectsBase(),
|
||||
_paneLister = paneLister ?? _tmuxPaneLister,
|
||||
_pollInterval = pollInterval;
|
||||
|
||||
final String workspacePath;
|
||||
final DaemonBus _events;
|
||||
final MessageBus _messages;
|
||||
final String _teamsBase;
|
||||
final String _projectsBase;
|
||||
final PaneLister _paneLister;
|
||||
final Duration _pollInterval;
|
||||
|
||||
Timer? _timer;
|
||||
bool _disposed = false;
|
||||
final Map<String, _LiveMember> _live = {};
|
||||
|
||||
static String _defaultTeamsBase() {
|
||||
final home = Platform.environment['HOME'] ?? '';
|
||||
return home.isNotEmpty ? '$home/.claude/teams' : '.claude/teams';
|
||||
}
|
||||
|
||||
static String _defaultProjectsBase() {
|
||||
final home = Platform.environment['HOME'] ?? '';
|
||||
return home.isNotEmpty ? '$home/.claude/projects' : '.claude/projects';
|
||||
}
|
||||
|
||||
static Future<Set<String>> _tmuxPaneLister() async {
|
||||
try {
|
||||
final r = await Process.run('tmux', ['-L', 'clide', 'list-panes', '-a', '-F', '#{pane_id}']);
|
||||
if (r.exitCode != 0) return const {};
|
||||
return (r.stdout as String).split('\n').map((s) => s.trim()).where((s) => s.isNotEmpty).toSet();
|
||||
} catch (_) {
|
||||
return const {};
|
||||
}
|
||||
}
|
||||
|
||||
/// Begin polling.
|
||||
void start() => _scheduleNext();
|
||||
|
||||
void _scheduleNext() {
|
||||
_timer = Timer(_pollInterval, () async {
|
||||
if (_disposed) return;
|
||||
await tick();
|
||||
if (!_disposed) _scheduleNext();
|
||||
});
|
||||
}
|
||||
|
||||
/// One poll cycle (public for tests). Diffs the config roster against the
|
||||
/// live panes and emits born/died.
|
||||
Future<void> tick() async {
|
||||
final config = await discoverTeam(workspacePath, teamsBase: _teamsBase);
|
||||
if (config == null) {
|
||||
await _killAll();
|
||||
return;
|
||||
}
|
||||
final livePanes = await _paneLister();
|
||||
final configIds = <String>{};
|
||||
|
||||
for (final m in config.teammates) {
|
||||
configIds.add(m.agentId);
|
||||
final paneLive = livePanes.contains(m.tmuxPaneId);
|
||||
final tracked = _live.containsKey(m.agentId);
|
||||
if (paneLive && !tracked) {
|
||||
await _born(config, m);
|
||||
} else if (!paneLive && tracked) {
|
||||
await _died(m.agentId);
|
||||
}
|
||||
}
|
||||
|
||||
// A member dropped from the config (team reshaped) also counts as died.
|
||||
for (final id in _live.keys.toList()) {
|
||||
if (!configIds.contains(id)) await _died(id);
|
||||
}
|
||||
}
|
||||
|
||||
Future<void> _born(TeamConfig config, TeamMember m) async {
|
||||
final path = await _resolveTranscript(config, m);
|
||||
TranscriptPublisher? pub;
|
||||
if (path != null) {
|
||||
pub = TranscriptPublisher(
|
||||
messages: _messages,
|
||||
reader: TranscriptReader(m.cwd ?? workspacePath, file: path, projectsBase: _projectsBase),
|
||||
channel: ClaudeConversation.teammateChannel(m.agentId),
|
||||
);
|
||||
}
|
||||
_live[m.agentId] = _LiveMember(m, config.team, pub);
|
||||
_events.emit(TeamMemberBorn(
|
||||
team: config.team,
|
||||
agentId: m.agentId,
|
||||
name: m.name,
|
||||
agentType: m.agentType,
|
||||
paneId: m.tmuxPaneId,
|
||||
model: m.model,
|
||||
color: m.color,
|
||||
cwd: m.cwd,
|
||||
transcriptPath: path,
|
||||
));
|
||||
}
|
||||
|
||||
Future<void> _died(String agentId) async {
|
||||
final live = _live.remove(agentId);
|
||||
if (live == null) return;
|
||||
await live.publisher?.dispose();
|
||||
_events.emit(TeamMemberDied(team: live.team, agentId: agentId, paneId: live.member.tmuxPaneId));
|
||||
}
|
||||
|
||||
Future<void> _killAll() async {
|
||||
for (final id in _live.keys.toList()) {
|
||||
await _died(id);
|
||||
}
|
||||
}
|
||||
|
||||
/// Best-effort join of [member] to its subagent transcript file. See the
|
||||
/// library doc — this is the drift-prone part. Returns null if no
|
||||
/// transcript can be resolved.
|
||||
Future<String?> _resolveTranscript(TeamConfig config, TeamMember member) async {
|
||||
final cwd = member.cwd;
|
||||
if (cwd == null || config.leadSessionId.isEmpty) return null;
|
||||
final munged = cwd.replaceAll('/', '-');
|
||||
final subDir = Directory('$_projectsBase/$munged/${config.leadSessionId}/subagents');
|
||||
if (!await subDir.exists()) return null;
|
||||
|
||||
final files = <File>[];
|
||||
await for (final e in subDir.list()) {
|
||||
if (e is File && e.path.endsWith('.jsonl') && !e.path.contains('compact')) {
|
||||
files.add(e);
|
||||
}
|
||||
}
|
||||
if (files.isEmpty) return null;
|
||||
|
||||
// Clean join: a sibling `.meta.json` whose agentType matches.
|
||||
for (final f in files) {
|
||||
final metaPath = '${f.path.substring(0, f.path.length - '.jsonl'.length)}.meta.json';
|
||||
final meta = File(metaPath);
|
||||
if (!await meta.exists()) continue;
|
||||
try {
|
||||
final m = jsonDecode(await meta.readAsString());
|
||||
if (m is Map && m['agentType'] == member.agentType) return f.path;
|
||||
} catch (_) {
|
||||
// ignore malformed meta
|
||||
}
|
||||
}
|
||||
|
||||
// Fallback: zip teammates-by-joinedAt against files-by-mtime.
|
||||
final teammates = [...config.teammates]..sort((a, b) => (a.joinedAt ?? 0).compareTo(b.joinedAt ?? 0));
|
||||
final idx = teammates.indexWhere((m) => m.agentId == member.agentId);
|
||||
if (idx < 0) return null;
|
||||
final stats = <(File, DateTime)>[];
|
||||
for (final f in files) {
|
||||
stats.add((f, (await f.stat()).modified));
|
||||
}
|
||||
stats.sort((a, b) => a.$2.compareTo(b.$2));
|
||||
return idx < stats.length ? stats[idx].$1.path : null;
|
||||
}
|
||||
|
||||
Future<void> dispose() async {
|
||||
_disposed = true;
|
||||
_timer?.cancel();
|
||||
_timer = null;
|
||||
await _killAll();
|
||||
}
|
||||
}
|
||||
@@ -187,10 +187,12 @@ class TranscriptReader {
|
||||
void Function(String)? onWarn,
|
||||
String? projectsBase,
|
||||
int? initialTailBytes,
|
||||
String? file,
|
||||
}) : _pollInterval = pollInterval,
|
||||
_onWarn = onWarn ?? _defaultWarn,
|
||||
_projectsBase = projectsBase ?? _defaultProjectsBase(),
|
||||
_initialTailBytes = initialTailBytes ?? _defaultInitialTailBytes;
|
||||
_initialTailBytes = initialTailBytes ?? _defaultInitialTailBytes,
|
||||
_explicitFile = file;
|
||||
|
||||
final String workspacePath;
|
||||
final Duration _pollInterval;
|
||||
@@ -205,6 +207,11 @@ class TranscriptReader {
|
||||
/// temp directory instead of the user's home.
|
||||
final String _projectsBase;
|
||||
|
||||
/// When set, tail this exact file instead of discovering the newest
|
||||
/// `.jsonl` in the munged dir. Used for teammate subagent transcripts
|
||||
/// (T-139), whose path the team observer resolves explicitly.
|
||||
final String? _explicitFile;
|
||||
|
||||
static String _defaultProjectsBase() {
|
||||
final home = Platform.environment['HOME'] ?? '';
|
||||
return home.isNotEmpty ? '$home/.claude/projects' : '.claude/projects';
|
||||
@@ -287,10 +294,9 @@ class TranscriptReader {
|
||||
}
|
||||
|
||||
Future<void> _tick(StreamController<ConversationItem> controller) async {
|
||||
final dir = _mungedDir();
|
||||
|
||||
// Discover or refresh the active session file.
|
||||
final newest = await _newestJsonl(dir);
|
||||
// A teammate reader tails one fixed file; otherwise discover the
|
||||
// newest session `.jsonl` in the munged dir.
|
||||
final newest = _explicitFile ?? await _newestJsonl(_mungedDir());
|
||||
if (newest == null) return;
|
||||
|
||||
if (newest != _currentPath) {
|
||||
|
||||
@@ -106,3 +106,74 @@ class DaemonEvent extends ClideEvent {
|
||||
@override
|
||||
Map<String, Object?> payload() => {'ts': ts.toIso8601String(), ...data};
|
||||
}
|
||||
|
||||
/// A Claude Code tmux teammate appeared (its pane is live). Identity is
|
||||
/// taken from the team config, so it is reliable regardless of transcript
|
||||
/// drift (T-139, D-75). [transcriptPath] is the best-effort resolved
|
||||
/// subagent transcript, or null if it could not be joined yet.
|
||||
class TeamMemberBorn extends ClideEvent {
|
||||
const TeamMemberBorn({
|
||||
required this.team,
|
||||
required this.agentId,
|
||||
required this.name,
|
||||
required this.agentType,
|
||||
required this.paneId,
|
||||
this.model,
|
||||
this.color,
|
||||
this.cwd,
|
||||
this.transcriptPath,
|
||||
});
|
||||
|
||||
/// Team name (the `~/.claude/teams/<team>` directory).
|
||||
final String team;
|
||||
|
||||
/// Config agent id (`<name>@<team>`).
|
||||
final String agentId;
|
||||
final String name;
|
||||
final String agentType;
|
||||
|
||||
/// tmux pane id (`%N`).
|
||||
final String paneId;
|
||||
final String? model;
|
||||
final String? color;
|
||||
final String? cwd;
|
||||
final String? transcriptPath;
|
||||
|
||||
@override
|
||||
String get subsystem => 'team';
|
||||
@override
|
||||
String get kind => 'member-born';
|
||||
@override
|
||||
Map<String, Object?> payload() => {
|
||||
'team': team,
|
||||
'agentId': agentId,
|
||||
'name': name,
|
||||
'agentType': agentType,
|
||||
'paneId': paneId,
|
||||
if (model != null) 'model': model,
|
||||
if (color != null) 'color': color,
|
||||
if (cwd != null) 'cwd': cwd,
|
||||
if (transcriptPath != null) 'transcriptPath': transcriptPath,
|
||||
};
|
||||
}
|
||||
|
||||
/// A Claude Code tmux teammate's pane went away (it exited or the team
|
||||
/// dissolved) — T-139.
|
||||
class TeamMemberDied extends ClideEvent {
|
||||
const TeamMemberDied({
|
||||
required this.team,
|
||||
required this.agentId,
|
||||
required this.paneId,
|
||||
});
|
||||
|
||||
final String team;
|
||||
final String agentId;
|
||||
final String paneId;
|
||||
|
||||
@override
|
||||
String get subsystem => 'team';
|
||||
@override
|
||||
String get kind => 'member-died';
|
||||
@override
|
||||
Map<String, Object?> payload() => {'team': team, 'agentId': agentId, 'paneId': paneId};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user