#!/usr/bin/env python3 """Session operations dispatcher — project updates, status summary, and session lifecycle.""" import argparse import json import os import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from flydocs_api import ( get_client, output_json, fail, resolve_text_input, find_project_root, take_last_relay_error, RelayError, DEFAULT_LIST_LIMIT, ) from atomic_io import atomic_write_text, atomic_write_json # --------------------------------------------------------------------------- # Session directory resolution (mirrors hook logic for consistency) # --------------------------------------------------------------------------- def _find_workspace_root_from(start: Path) -> Path | None: """Walk up to find .flydocs-workspace.json.""" current = start while current != current.parent: if (current / '.flydocs-workspace.json').is_file(): return current current = current.parent return None def _resolve_session_dir(project_root: Path) -> Path: """Resolve workspace-scoped session directory.""" config_file = project_root / '.flydocs' / 'config.json' workspace_id = None if config_file.exists(): try: config = json.loads(config_file.read_text()) workspace_id = config.get('workspaceId') except (json.JSONDecodeError, OSError): pass scope = workspace_id if workspace_id else 'default' return project_root / '.flydocs' / 'session' / scope def _atomic_write(path: Path, content: str) -> None: """Write content atomically via write-then-replace (FLY-718, FLY-1293). Delegates to the shared helper. The local implementation used ``Path.rename``, which raises on Windows when the target already exists — and every session write targets a file that already exists. """ atomic_write_text(path, content) def _atomic_write_json(path: Path, data: object) -> None: """Atomically write JSON data with validation (FLY-718).""" atomic_write_json(path, data) def _read_json_safe(path: Path) -> dict | None: """Read and parse JSON, returning None on any error.""" if not path.exists(): return None try: return json.loads(path.read_text()) except (json.JSONDecodeError, OSError): return None def _read_text_safe(path: Path) -> str | None: """Read text file, returning None on any error.""" if not path.exists(): return None try: return path.read_text().strip() except (OSError, IOError): return None def _delete_safe(path: Path) -> bool: """Delete a file if it exists. Returns True if deleted.""" if not path.exists(): return False try: path.unlink() return True except OSError: return False # --------------------------------------------------------------------------- # start-context: gather all session start data in one call # --------------------------------------------------------------------------- def cmd_start_context(args): """Gather all data needed for /start-session in one deterministic call.""" project_root = find_project_root() config_file = project_root / '.flydocs' / 'config.json' config = _read_json_safe(config_file) or {} session_dir = _resolve_session_dir(project_root) workspace_root = _find_workspace_root_from(Path.cwd()) result = { 'projectRoot': str(project_root), 'isMultiRepo': workspace_root is not None, } # --- Identity --- identity = None me_local = project_root / '.flydocs' / 'me.json' me_global = Path.home() / '.flydocs' / 'me.json' if me_local.exists(): identity = _read_json_safe(me_local) elif me_global.exists(): identity = _read_json_safe(me_global) result['identity'] = { 'name': identity.get('name') if identity else None, 'email': identity.get('email') if identity else None, } # --- Workspace --- workspace_name = config.get('workspaceName') if not workspace_name: cache = _read_json_safe(project_root / '.flydocs' / 'validation-cache.json') if cache: workspace_name = cache.get('workspace', {}).get('name') result['workspace'] = { 'name': workspace_name, 'tier': config.get('tier', 'local'), 'onboardComplete': config.get('onboardComplete', False), 'setupComplete': config.get('setupComplete', False), } # --- Active board/project --- (ADR-011: direct top-level reads) active_contexts = config.get('activeContexts', []) active_project_id = config.get('activeProjectId') # Migration: read old activeProjects[0] if activeProjectId not set if not active_project_id: ap = config.get('activeProjects', []) if ap: active_project_id = ap[0] board = None if active_contexts: primary = active_contexts[0] board = { 'name': primary.get('name') or primary.get('boardName'), 'type': primary.get('boardType'), 'sprintId': primary.get('sprintId'), 'sprintName': primary.get('sprintName'), } result['activeBoard'] = board result['hasActiveProject'] = bool(active_project_id) or bool(active_contexts) # --- Sprint --- sprint_id = None if board and board.get('type') == 'scrum': sprint_id = board.get('sprintId') if not sprint_id: sprint_id = config.get('activeSprintId') result['activeSprintId'] = sprint_id # --- Last session (the critical handoff artifact) --- last_summary = _read_json_safe(session_dir / 'last-summary.json') result['lastSession'] = last_summary # --- Active issue (from focus files) --- focus_text = _read_text_safe(session_dir / 'focus.md') active_issue = None if focus_text: match = re.search(r'[A-Z]+-[0-9]+', focus_text) if match: status = _read_text_safe(session_dir / 'status') ac_text = _read_text_safe(session_dir / 'acceptance-criteria.md') or '' total = len(re.findall(r'^\s*-\s*\[', ac_text, re.MULTILINE)) done = len(re.findall(r'^\s*-\s*\[x\]', ac_text, re.MULTILINE | re.IGNORECASE)) active_issue = { 'id': match.group(0), 'status': status, 'acTotal': total, 'acDone': done, } result['activeIssue'] = active_issue # --- Multi-repo: aggregate sibling session summaries --- if workspace_root: ws_data = _read_json_safe(workspace_root / '.flydocs-workspace.json') or {} repos = [] sibling_sessions = [] for name, entry in ws_data.get('repos', {}).items(): repo_path = (workspace_root / entry.get('path', '')).resolve() repos.append({ 'name': name, 'path': str(repo_path), 'purpose': entry.get('purpose', ''), }) # Read each repo's last-summary sibling_sd = _resolve_session_dir(repo_path) sibling_summary = _read_json_safe(sibling_sd / 'last-summary.json') if sibling_summary: sibling_sessions.append({ 'repo': name, 'summary': sibling_summary, }) result['repos'] = repos result['siblingSessionSummaries'] = sibling_sessions output_json(result) # --------------------------------------------------------------------------- # list-issues: board-scoped issue fetch with automatic fallback # --------------------------------------------------------------------------- def cmd_list_issues(args): """Fetch the user's issues with automatic board/sprint scoping. Reads config to determine the active board type and sprint, applies the correct filter, and falls back to unfiltered if the scoped query returns zero results or if no board/sprint is configured. """ project_root = find_project_root() config_file = project_root / '.flydocs' / 'config.json' config = _read_json_safe(config_file) or {} limit = str(args.limit) # Resolve the issues.py script path issues_script = Path(__file__).parent / 'issues.py' if not issues_script.exists(): fail(f'issues.py not found at {issues_script}') # Determine scoping from config (ADR-011: direct top-level reads) active_contexts = config.get('activeContexts', []) sprint_id = config.get('activeSprintId') board_type = None if active_contexts: primary = active_contexts[0] board_type = primary.get('boardType') ctx_sprint = primary.get('sprintId') if ctx_sprint: sprint_id = ctx_sprint # Build the scoped command base_cmd = [sys.executable, str(issues_script), 'list', '--active', '--mine', '--limit', limit] scoped_cmd = list(base_cmd) filter_used = 'none' if board_type == 'scrum' and sprint_id: scoped_cmd.extend(['--sprint', sprint_id]) filter_used = 'sprint' elif board_type == 'kanban': scoped_cmd.extend(['--board', 'active']) filter_used = 'board' elif sprint_id: # v2 compat: no board type info but sprint is set scoped_cmd.extend(['--sprint', sprint_id]) filter_used = 'sprint' # Try scoped query first issues = [] try: result = subprocess.run( scoped_cmd, capture_output=True, text=True, timeout=30, cwd=str(project_root) ) if result.returncode == 0: issues = json.loads(result.stdout) except (subprocess.TimeoutExpired, subprocess.SubprocessError, json.JSONDecodeError): pass # Fallback: if scoped returned zero or if no filter was applied, try unfiltered fallback_used = False if not issues and filter_used != 'none': try: result = subprocess.run( base_cmd, capture_output=True, text=True, timeout=30, cwd=str(project_root) ) if result.returncode == 0: issues = json.loads(result.stdout) if issues: fallback_used = True except (subprocess.TimeoutExpired, subprocess.SubprocessError, json.JSONDecodeError): pass elif not issues and filter_used == 'none': # No filter available — the base_cmd IS the only query try: result = subprocess.run( base_cmd, capture_output=True, text=True, timeout=30, cwd=str(project_root) ) if result.returncode == 0: issues = json.loads(result.stdout) except (subprocess.TimeoutExpired, subprocess.SubprocessError, json.JSONDecodeError): pass output_json({ 'issues': issues, 'count': len(issues), 'filter': filter_used, 'fallbackUsed': fallback_used, }) # --------------------------------------------------------------------------- # wrap: deterministic session wrap with cleanup # --------------------------------------------------------------------------- # FLY-990: The project update posted at session wrap must be the filled # session-wrap template. These four section headers are mandatory; "Notes" # is optional. "Blockers" also matches "Blockers & open questions". # # FLY-1411: the section list is now one field of a *named, versioned template # resource*. The relay validates the same list server-side # (`convex/lib/sessionWrapTemplate.ts`), and two enforcement points sharing a # contract need a name and a version to disagree about — an anonymous tuple # would have left "which sections?" answerable only by reading two files. WRAP_TEMPLATE = { "id": "session-wrap", "version": 1, "required_sections": ("Accomplished", "Next up", "Blockers", "Progress"), } # The historical name, kept as an alias: the hooks, the MCP shaping layer and # the enforcement suite all read it, and the constant is the same object. REQUIRED_WRAP_SECTIONS = WRAP_TEMPLATE["required_sections"] _WRAP_TEMPLATE_PATH = ( ".claude/skills/flydocs-workflow/templates/session/session-wrap.md" ) # --- SessionUpdate v1 (FLY-1411, phase 10 spec §4.1) ----------------------- SESSION_UPDATE_SCHEMA_VERSION = 1 #: Where the local record lands, and what it is never pruned by. STREAM_FILENAME = "stream.jsonl" #: Rotated generations: stream.1.jsonl … stream.3.jsonl (oldest last). STREAM_ROTATED_PATTERN = re.compile(r"^stream\.\d+\.jsonl$") STREAM_MAX_BYTES = 5 * 1024 * 1024 STREAM_ROTATION_KEEP = 3 VISIBILITY_CHOICES = ("private", "team", "org", "external") DEFAULT_VISIBILITY = "team" # FLY-1498: the two human-written fields of the record. Every destination — the # Activity timeline, the modal, the Slack parent, the email digest — opens with # one of them, and before this they were *derived* (first heading, first # paragraph), which is a fallback dressed up as a design. # # The bounds are the relay's, and the relay measures them with JavaScript's # `String.length` — UTF-16 code units, not code points. A summary of emoji is # half as long to Python's `len()` as it is to the validator, so everything # here counts the unit the server counts. SESSION_TITLE_MAX = 120 SESSION_SUMMARY_MAX = 600 def _utf16_len(text: str) -> int: """Length in UTF-16 code units — what the relay's `String.length` counts.""" return len(text.encode("utf-16-le")) // 2 def _truncate_utf16(text: str, limit: int) -> str: """`text` cut to at most `limit` UTF-16 code units. Cuts between characters, never between the halves of a surrogate pair: an astral character is 2 units, so one that does not fit whole is dropped whole rather than turned into an unpaired surrogate the server would refuse to decode. """ if _utf16_len(text) <= limit: return text used = 0 end = 0 for index, char in enumerate(text): width = 2 if ord(char) > 0xFFFF else 1 if used + width > limit: break used += width end = index + 1 return text[:end] def normalize_title(raw: str | None) -> str | None: """The envelope `title` for a `--title` argument, or None to omit the key. One line, always: the route refuses a title containing a line break, and a title read out of a file arrives with a trailing `\\r\\n` through no fault of the person who wrote it. Every run of whitespace — breaks, tabs and the Unicode line separators included — collapses to a single space rather than being argued about over HTTP. Omission is the empty case. `""` and `null` are both refused by the mutation, and neither says anything a missing key does not. """ if not raw: return None # `\s` on a str pattern is the Unicode class: it already covers `\r\n`, # `\v`, `\f`, NBSP and U+2028/U+2029, which is every character the # route's single-line check could trip on. collapsed = re.sub(r"\s+", " ", raw).strip() if not collapsed: return None # `rstrip` because unit 120 can land on a word boundary, and the route # trims what it stores: without this the stored title and the one the wrap # echoes back as "sent" differ by a trailing space. It cannot empty the # string — index 0 is non-space after the collapse above. return _truncate_utf16(collapsed, SESSION_TITLE_MAX).rstrip() def normalize_summary(raw: str | None) -> str | None: """The envelope `summary` for a `--notes` argument, or None to omit it. `--notes` is the summary. It always was in intent — free-form prose about the session — and it was only ever used to label the graph node, so this gives it the destination it was written for without changing what a caller passes. Over the bound it is cut at the last sentence end inside the limit, so a reader gets a whole thought rather than a word broken in half. A block with no sentence end in its second half is cut hard: a "boundary" three characters in is not one. """ if not raw: return None text = raw.strip() if not text: return None if _utf16_len(text) <= SESSION_SUMMARY_MAX: return text cut = _truncate_utf16(text, SESSION_SUMMARY_MAX) ends = list(re.finditer(r"[.!?](?=\s|$)", cut)) if ends and ends[-1].end() > len(cut) // 2: return cut[: ends[-1].end()].strip() return cut.strip() def validate_wrap_body(body: str) -> list[str]: """Return required section headers missing from a wrap body. Recognizes both Markdown (`## Heading`) and bold (`**Heading**`) header styles, matched case-insensitively. A section counts as present when its name appears in any header line, so "Blockers" is satisfied by a "Blockers & open questions" header. Returns [] when all required sections are present. """ if not body: return list(REQUIRED_WRAP_SECTIONS) headers: list[str] = [] for line in body.splitlines(): stripped = line.strip() md = re.match(r'^#{1,6}\s+(.*)', stripped) if md: headers.append(md.group(1)) continue bold = re.match(r'^\*\*(.+?)\*\*', stripped) if bold: headers.append(bold.group(1)) haystack = '\n'.join(headers).lower() return [s for s in REQUIRED_WRAP_SECTIONS if s.lower() not in haystack] # --------------------------------------------------------------------------- # The session record (FLY-1411) # # The wrap body stops being the artifact and becomes one field of it. The # envelope composed here is what the relay stores as a `sessionUpdates` row and # what every destination — the provider project update included — renders from; # the same bytes are appended to `stream.jsonl`, which is the whole feature on # the local tier and the offline copy on the cloud one. # --------------------------------------------------------------------------- # Every file the session directory holds as STATE — written during a session, # meaningless after it, and cleared by wrap. Named once because the cost of # forgetting one is a file that outlives its session: `status-ref` was left off # this list and a wrapped issue came back as the next session's attribution # subject (FLY-1356). `mirror-mismatch-notice` records what `auto-approve.py` # has already reported about the state of the other four (FLY-1471), so it # expires with them. # # The stream is deliberately absent: it is history, not state (FLY-1411). SESSION_STATE_FILES: tuple[str, ...] = ( 'focus.md', 'status', 'status-ref', 'acceptance-criteria.md', 'mirror-mismatch-notice', ) def is_stream_file(name: str) -> bool: """True for the record stream and its rotated generations. The one place that answers "may this file be deleted?" — every cleanup, archive or prune path in this module consults it rather than carrying its own spelling of the filename. """ return name == STREAM_FILENAME or bool(STREAM_ROTATED_PATTERN.match(name)) def _rotate_stream(path: Path) -> None: """Roll `stream.jsonl` over at the size cap, keeping three generations. Rotation is a rename, never a truncate: the bytes already written are the record, and a log that rewrites itself is a log that can lose one. """ try: if not path.exists() or path.stat().st_size < STREAM_MAX_BYTES: return except OSError: return # Two wraps can reach the cap together, and the loser finds every file the # winner already renamed. That is a race over *bookkeeping*, and it must # never surface as a failed wrap: by the time rotation runs the relay has # accepted the record, so an exception here would report a wrap that did # not happen and invite a retry that duplicates it. session_dir = path.parent oldest = session_dir / f"stream.{STREAM_ROTATION_KEEP}.jsonl" _delete_safe(oldest) for generation in range(STREAM_ROTATION_KEEP - 1, 0, -1): source = session_dir / f"stream.{generation}.jsonl" try: source.replace(session_dir / f"stream.{generation + 1}.jsonl") except (FileNotFoundError, OSError): continue try: path.replace(session_dir / "stream.1.jsonl") except (FileNotFoundError, OSError): # The other wrap rotated it; this one appends to the fresh file. return def append_session_stream(session_dir: Path, envelope: dict, relay: dict, window_source: str) -> Path: """Append one envelope to the local record stream. Returns its path. One `os.write` of the whole encoded line to an `O_APPEND` descriptor. A record is comfortably larger than the 8 KiB a text-mode buffer flushes in, so `open('a').write()` can interleave two concurrent wraps mid-line and leave two corrupt records where there should be two good ones; a single write to an append-only descriptor is atomic for the sizes involved and is what makes "a reader never sees half a record" true rather than likely. """ path = session_dir / STREAM_FILENAME _rotate_stream(path) record = dict(envelope) record["local"] = { "appendedAt": datetime.now(timezone.utc).isoformat(), "windowSource": window_source, "relay": relay, } line = (json.dumps(record, ensure_ascii=False) + "\n").encode("utf-8") fd = os.open(path, os.O_WRONLY | os.O_APPEND | os.O_CREAT, 0o644) try: os.write(fd, line) finally: os.close(fd) return path def _detect_actor(project_root: Path) -> dict: """Identify the agent behind this wrap from the environment. Nothing in the workspace records a harness today, so this reads the variables the harnesses themselves set: `CLAUDECODE` / `CLAUDE_*` (Claude Code, already the hooks' signal for a project dir), `CURSOR_*`, `CODEX_*`. `FLYDOCS_HARNESS` overrides everything — it is the seam for a harness that advertises nothing, and the only way a version reaches `harness` until one of them exposes theirs. """ env = os.environ agent = None if env.get("CLAUDECODE") or env.get("CLAUDE_CODE_ENTRYPOINT") \ or env.get("CLAUDE_PROJECT_DIR"): agent = "claude-code" elif env.get("CURSOR_TRACE_ID") or env.get("CURSOR_AGENT"): agent = "cursor" elif any(name.startswith("CODEX_") for name in env): agent = "codex" harness = env.get("FLYDOCS_HARNESS") display_name = None for me_path in (project_root / '.flydocs' / 'me.json', Path.home() / '.flydocs' / 'me.json'): me = _read_json_safe(me_path) if me: display_name = me.get('displayName') or me.get('name') if display_name: break actor: dict = {} if agent: actor['agent'] = agent if harness: actor['harness'] = harness if display_name: actor['displayName'] = display_name return actor def _resolve_repo(project_root: Path, config: dict, client: object = None) -> str: """The repo slug to record, and the one the request will claim to be from. The relay's own slug wins when there is one. `X-Repo` is set from it on every request, and the route refuses a body whose `repo` disagrees with the header — so the two have to be one value, not two independently-plausible ones. Config's `repoSlug` is next — the body's fallback when the relay has no slug; the header is simply absent then, and the route only checks agreement when both are present — and the directory name is the last resort: right often enough to beat omitting the field the record is indexed by. """ relay = getattr(client, '_relay', None) if client is not None else None detected = getattr(relay, 'repo_slug', None) if isinstance(detected, str) and detected: return detected slug = config.get('repoSlug') if isinstance(slug, str) and slug: return slug return project_root.name def _sequences_in_graph(project_root: Path, session_date: str) -> list[int]: """Sequence numbers already used by session nodes for this date. The id shape comes from `graph_session`, which creates the nodes — one declaration, so the reader here and the writer there cannot drift. """ try: from graph_session import SESSION_ID_PATTERN from graph_utils import load_graph nodes = load_graph(project_root).get('nodes', {}) except Exception: return [] used = [] for node_id in nodes: if not node_id.startswith('session:'): continue match = SESSION_ID_PATTERN.match(node_id.split(':', 1)[1]) if match and match.group(1) == session_date: used.append(int(match.group(2)) if match.group(2) else 0) return used def _sequences_in_stream(session_dir: Path, session_date: str) -> list[int]: """Sequence numbers already recorded for this date, across generations. Scanned with a regex rather than parsed: the only thing wanted from a 5 MB log is which ids it already used, and every line is a `sessionId` away from the answer. """ pattern = re.compile( r'"sessionId"\s*:\s*"' + re.escape(session_date) + r'(?:-(\d+))?"' ) used = [] for name in [STREAM_FILENAME] + [ f'stream.{n}.jsonl' for n in range(1, STREAM_ROTATION_KEEP + 1) ]: path = session_dir / name if not path.exists(): continue try: text = path.read_text(encoding='utf-8', errors='replace') except OSError: continue for match in pattern.finditer(text): used.append(int(match.group(1)) if match.group(1) else 0) return used def resolve_session_id(project_root: Path, session_dir: Path, session_date: str) -> str: """The id for this session — `YYYY-MM-DD-`, unique within the day. The sequence is always present, and it is derived from **both** the graph and the record stream. Deriving it from the graph alone was wrong on the primary path: the graph node is only written when the wrap has issues, and `session_wrap` over MCP sends no notes, so nothing advanced the sequence and two wraps on the same day claimed the same id — which the relay treats as one session, and the second record would have superseded the first. The stream is the more reliable of the two here (every posted wrap appends to it), and the graph keeps a wrap that ran before this scheme existed — including the un-suffixed `session:2026-08-29` — from being re-used. """ used = _sequences_in_graph(project_root, session_date) \ + _sequences_in_stream(session_dir, session_date) return f"{session_date}-{max(used, default=0) + 1}" def narrative_summary(body: str | None) -> str | None: """A one-line summary of a wrap body: its first heading and first bullet. What the graph node is labelled with when the caller passed no `notes`. """ if not body: return None lines = body.splitlines() for index, line in enumerate(lines): heading = re.match(r'^#{1,6}\s+(.*)', line.strip()) if not heading: continue for follow in lines[index + 1:]: text = re.sub(r'^[-*]\s+', '', follow.strip()) if text and not text.startswith('#'): return f"{heading.group(1).strip()} — {text}" return heading.group(1).strip() return None def _parse_started_at(raw: str | None) -> int | None: """Read `--started-at` as unix ms, accepting epoch ms or an ISO timestamp.""" if not raw: return None text = raw.strip() if re.fullmatch(r'\d+', text): return int(text) try: parsed = datetime.fromisoformat(text.replace('Z', '+00:00')) except ValueError: return None if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=timezone.utc) return int(parsed.timestamp() * 1000) def compose_session_update(*, repo: str, session_id: str, actor: dict, started_at: int, ended_at: int, health: str, issues: list, pending: list, blockers: list, narrative: str, visibility: str, cli_version: str | None = None, title: str | None = None, summary: str | None = None) -> dict: """Build the SessionUpdate v1 client envelope (spec §4.1). Pure: every input is already resolved. `narrative.markdown` is the posted body byte for byte — the record is a superset of what the provider update said, never a re-rendering of it, which is what lets a destination be swapped without the record changing. `shipped`, `decisions` and `metrics` are the fields v1 knowingly leaves empty: pull requests and decisions come from the Phase 8 provenance join, numbers from Phase 7, and neither is available to a dispatcher script. FLY-1498: `title` and `summary` are optional and **omitted when absent**. Both arrive already normalized (`normalize_title` / `normalize_summary`), so this stays the pure composition it was — and a caller that has neither produces the same bytes it produced before the fields existed. """ envelope: dict = { 'schemaVersion': SESSION_UPDATE_SCHEMA_VERSION, 'repo': repo, 'sessionId': session_id, 'actor': actor, 'window': {'startedAt': started_at, 'endedAt': ended_at}, 'health': health, 'issues': issues, 'shipped': [], 'decisions': [], 'pending': pending, 'blockers': blockers, 'narrative': {'markdown': narrative}, 'visibility': visibility, 'provenance': {'source': 'cli'}, } if cli_version: envelope['provenance']['cliVersion'] = cli_version # Set, never blanked. The mutation refuses `null` and an empty string, and # a key that is not there is the only spelling of "the writer gave none" # that every reader — route, resolver, `flydocs stream` — already handles. if title: envelope['title'] = title if summary: envelope['summary'] = summary return envelope def _clean_texts(values: object) -> list[str]: """The non-blank strings in a list argument, stripped.""" if not isinstance(values, (list, tuple)): return [] return [v.strip() for v in values if isinstance(v, str) and v.strip()] def _findings_sentence(err: RelayError) -> str: """The server's per-item findings, rendered for a terminal. Both shapes the relay uses: `{section, status}` for a wrap-body rejection and `{field, message}` for an envelope one. """ findings = (err.body or {}).get('findings') or [] named = [] for finding in findings: if not isinstance(finding, dict): continue subject = finding.get('section') or finding.get('field') detail = finding.get('status') or finding.get('message') if subject: named.append(f"{subject} ({detail})" if detail else str(subject)) return f" Findings: {', '.join(named)}." if named else "" def _is_deterministic_rejection(err: RelayError) -> bool: """True for a 4xx the server will answer the same way next time. 409 and 429 are the two that will not: an operation still in flight and a throttle are both "later", and `_request` has already spent its retries on them by the time one reaches here. """ return 400 <= err.status < 500 and err.status not in (409, 429) def _compose_issue_entries(issues: list, session_dir: Path) -> list[dict]: """`{ref}` per issue, carrying `statusTo` for the one the session tracked. `status`/`status-ref` are the pair the stop gate reads: the ref the session was attributed to and the status it was left in. Read before cleanup, which is the only reason the wrap knows a status at all. """ status_ref = _read_text_safe(session_dir / 'status-ref') status = _read_text_safe(session_dir / 'status') entries = [] for ref in issues: entry = {'ref': ref} if status and status_ref and status_ref.strip() == ref: entry['statusTo'] = status entries.append(entry) return entries def cmd_wrap(args): """Perform the full session wrap handoff deterministically.""" project_root = find_project_root() session_dir = _resolve_session_dir(project_root) session_dir.mkdir(parents=True, exist_ok=True) issues = args.issues or [] # FLY-1411: the route refuses a blank `pending[i]` or `blockers[].text`, # and it is right to — an empty bullet is not a pending item. A stray # `--pending ""` from a shell loop must not be what 400s a wrap, so the # blanks are dropped here rather than argued about over HTTP. pending = _clean_texts(args.pending) blockers = _clean_texts(args.blockers) notes = args.notes or '' # FLY-1498: the record's two human-written fields. Both are derived from # the flags alone — no clock, no graph sequence — so a respawned wrap under # one operation seed composes byte-identical strings and keys to the same # operation id rather than filing a second record. record_title = normalize_title(getattr(args, 'title', None)) record_summary = normalize_summary(notes) health = args.health posting = bool(health) # FLY-990: Resolve and validate the project-update body BEFORE any # mutation. The posted body must be the filled session-wrap template. # On the posting path (--health provided), a body missing any required # section is rejected outright — nothing is posted and session state # (focus.md/status/status-ref/acceptance-criteria.md) is left intact. posted_body = None if posting: body = resolve_text_input(text_arg=args.body, file_arg=args.body_file) if not body: # Auto-compose fallback (no template supplied). This does not # contain the required section headers, so the guard below # rejects it and points the user at the template. lines = [f'**Session Summary — {datetime.now().strftime("%Y-%m-%d")}**\n'] if issues: lines.append('**Issues worked:** ' + ', '.join(issues)) if pending: lines.append('**Pending:** ' + '; '.join(pending)) if blockers: lines.append('**Blockers:** ' + '; '.join(blockers)) if notes: lines.append('\n' + notes) body = '\n'.join(lines) missing = validate_wrap_body(body) if missing: fail( "Session-wrap body is missing required section(s): " + ", ".join(missing) + f". Fill {_WRAP_TEMPLATE_PATH} and pass via --body-file. " "Required sections: " + ", ".join(REQUIRED_WRAP_SECTIONS) + " (Notes optional). No project update was posted and " "session state was left intact." ) posted_body = body results = { 'success': True, 'actions': [], } # FLY-1411: compose the record before anything mutates, for the same reason # the body is validated there — `status`/`status-ref` are read here and # deleted in step 3, so the envelope has to be built while they still exist. envelope = None window_source = 'endedAt' client = None if posting: # The client is built before the envelope, not with the post: it owns # the repo slug the request will carry in `X-Repo`, and the route # refuses a body that names a different one. Building it here also # means a missing key or workspace fails before anything is written. client = get_client() config = _read_json_safe(project_root / '.flydocs' / 'config.json') or {} ended_at = int(datetime.now(timezone.utc).timestamp() * 1000) started_at = _parse_started_at(getattr(args, 'started_at', None)) if started_at is None: # No harness writes a session-start marker, so an unsupplied window # is a point in time rather than a guess at a duration. The stream # records which of the two this was; the server sees a valid # `startedAt == endedAt` either way. started_at = ended_at else: window_source = 'startedAt-flag' visibility = getattr(args, 'visibility', None) or DEFAULT_VISIBILITY envelope = compose_session_update( repo=_resolve_repo(project_root, config, client), session_id=resolve_session_id( project_root, session_dir, datetime.now().date().isoformat()), actor=_detect_actor(project_root), started_at=started_at, ended_at=ended_at, health=health, issues=_compose_issue_entries(issues, session_dir), pending=list(pending), blockers=[{'text': text} for text in blockers], narrative=posted_body, visibility=visibility, cli_version=os.environ.get('FLYDOCS_CLI_VERSION'), title=record_title, summary=record_summary, ) # 1. Write last-summary.json (the critical handoff artifact). The # validated posted body is the single source of truth for what was # posted, so persist it here. summary_data = { 'timestamp': datetime.now(timezone.utc).isoformat(), 'issues': issues, 'pending': pending, 'blockers': blockers, 'notes': notes, } if posted_body is not None: summary_data['postedBody'] = posted_body summary_file = session_dir / 'last-summary.json' try: _atomic_write_json(summary_file, summary_data) results['actions'].append({ 'action': 'write_summary', 'path': str(summary_file), 'success': True, }) except (OSError, ValueError) as e: results['actions'].append({ 'action': 'write_summary', 'success': False, 'error': str(e), }) results['success'] = False # 2. Store the record and post the project update (before cleanup, so # session state is only cleared on a successful post). # # FLY-1411: on the cloud tier the record is the write and the provider # update is a destination of it — one governed call replaces the direct # `/projects/update` post. On the local tier there is no relay and no # destination, so the file-store update stays exactly as it was and the # stream append below is the whole record. post_succeeded = False stream_ok = True relay_state = {'state': 'skipped'} if posting: project_id = args.project if not project_id and client.is_cloud: project_id = client.config.get('activeProjectId') # Migration: fall back to old activeProjects[0] if not project_id: ap = client.config.get('activeProjects', []) if ap: project_id = ap[0] if client.is_cloud: try: record = client.session_update_create(envelope, project_id=project_id) except Exception as e: # A server-side wrap rejection is the same failure as the local # validator's: the body is wrong, so nothing was stored, nothing # posted, nothing cleaned — and it exits rather than reporting a # half-success, exactly as the client-side guard does. # The exception carries the code when the caller asked for it # to be raised; `take_last_relay_error` is the fallback for a # path that rendered-and-exited before reaching here. err = e if isinstance(e, RelayError) else take_last_relay_error() if err is not None and err.code == 'WRAP_VALIDATION_FAILED': fail( "The relay refused the session-wrap body: " + err.message + _findings_sentence(err) + " Required sections: " + ", ".join(REQUIRED_WRAP_SECTIONS) + " (Notes optional). No session record was stored, no " "project update was posted and session state was left " "intact." ) if err is not None and _is_deterministic_rejection(err): # Every other 4xx is a decision the server will repeat: a # malformed envelope, an expired key, a workspace this repo # does not belong to. Appending it and keeping the state # invites a retry that appends a second identical record and # is refused identically — so it exits like the validator # does, with the server's own words. fail( f"The relay refused the session record ({err.code}): " + err.message + _findings_sentence(err) + " Nothing was stored, nothing was posted, nothing was " "appended and session state was left intact." ) relay_state = {'state': 'failed', 'error': str(e)} if err is not None: relay_state['code'] = err.code results['actions'].append({ 'action': 'session_update', 'success': False, 'error': str(e), }) results['success'] = False else: provider_update = record.get('providerUpdate') or {} relay_state = { 'state': 'accepted', 'id': record.get('id'), 'replayed': bool(record.get('replayed', False)), } stored = { 'action': 'session_update', 'success': True, 'id': record.get('id'), 'permalinkPath': record.get('permalinkPath'), 'replayed': bool(record.get('replayed', False)), 'providerUpdate': provider_update, } # FLY-1498: echo the two fields as they were *sent*. They are # normalized on the way out — collapsed, and cut at the # relay's bounds — and a caller that wrote a 700-character # summary should be able to see what its readers will get. # Reported only when present, so a wrap without them is the # same payload it was before the fields existed. if envelope.get('title'): stored['title'] = envelope['title'] if envelope.get('summary'): stored['summary'] = envelope['summary'] results['actions'].append(stored) post_succeeded = True # `posted: false` splits two different outcomes, and only one of # them is a problem. # # A `reason` is the route stating a fact about the destination: # `unsupported` (Jira and GitHub Issues have no project-update # concept) or `replayed` (this record already posted once). The # record landed, nothing is owed, the wrap is done. # # An `error` is a destination that could have taken the update # and did not. The record is stored either way, so the retry is # only the *post* — and posting the narrative is exactly what # `/projects/update` still does. Falling back to it here keeps # golden rule 5 ("the wrap posts the update") true through a # destination failure, instead of handing the user a wrapped # session with nothing in the team's feed. if provider_update.get('posted') is False and provider_update.get('error'): try: fallback = client.project_update( health, posted_body, project_id=project_id) results['actions'].append({ 'action': 'project_update', 'success': True, 'updateId': fallback.get('id'), 'viaFallback': True, 'destinationError': provider_update.get('error'), }) except Exception as post_error: # Both routes to the team's feed are shut. The record is # stored and appended, but the update nobody read is the # part of the wrap that did not happen — so the session # state stays for a retry and the MCP layer says so. results['actions'].append({ 'action': 'project_update', 'success': False, 'error': str(post_error), 'viaFallback': True, 'destinationError': provider_update.get('error'), }) results['success'] = False post_succeeded = False else: try: update_result = client.project_update( health, posted_body, project_id=project_id) post_succeeded = True results['actions'].append({ 'action': 'project_update', 'success': True, 'updateId': update_result.get('id'), }) except Exception as e: results['actions'].append({ 'action': 'project_update', 'success': False, 'error': str(e), }) results['success'] = False # The append happens once the outcome is known. On the cloud tier it # happens whatever that outcome was — the relay being unreachable must not # be the thing that loses the record (spec §9). On the local tier the # file-store write and the append are one act: appending after a failed # write would leave a record the retry appends a second copy of, and there # is no server-side identity to collapse the two. if posting and (client.is_cloud or post_succeeded): try: stream_path = append_session_stream( session_dir, envelope, relay_state, window_source) results['actions'].append({ 'action': 'session_stream', 'success': True, 'path': str(stream_path), 'relayState': relay_state['state'], }) except OSError as e: # Losing the local record silently is the failure mode phase 10 # exists to remove, so this joins the post in gating cleanup: the # session state stays, and the caller is told to run it again. stream_ok = False results['actions'].append({ 'action': 'session_stream', 'success': False, 'error': str(e), }) results['success'] = False # 3. Clean up session state files — only after a successful post, or # when not posting at all. Never when the guard blocked or the post # failed, so the handoff context survives for a retry. # The one condition: a wrap that did not land leaves everything alone — # the session state, and (below) the graph. wrap_landed = (not posting) or (post_succeeded and stream_ok) cleaned = [] if wrap_landed: for filename in SESSION_STATE_FILES: # FLY-1411: cleanup clears session *state*; the record stream is # not state, it is history, and it outlives every session in the # directory. The list above is explicit and contains no stream # file, so this guard exists for the next person who adds one — # a glob, a temp file, a rename — rather than for today. if is_stream_file(filename): continue if _delete_safe(session_dir / filename): cleaned.append(filename) results['actions'].append({ 'action': 'cleanup', 'filesRemoved': cleaned, }) # 4. Record in context graph (skip silently if not available) # # FLY-1411: `notes` is no longer required. It used to be, and # `session_wrap` over MCP sends none — so the primary path never recorded a # session node at all, and the temporal layer `graph_context.py` traverses # had a hole exactly where the tool surface was used. The summary falls back # to the wrap body's own first line. # # `wrap_landed` is the other half: a node written for a wrap that failed is # a node the retry writes again, so the graph would accumulate one session # per attempt at a session. if issues and wrap_landed: graph_script = Path(__file__).parent / 'graph_session.py' if graph_script.exists(): fallback = f"Session wrap ({health})" if health else "Session wrap" summary = notes or narrative_summary(posted_body) or fallback cmd = [ sys.executable, str(graph_script), '--summary', summary, ] if envelope is not None: # The record and the node name the same session. cmd.extend(['--session-id', envelope['sessionId']]) for issue_id in issues: cmd.extend(['--issue', issue_id]) try: proc = subprocess.run( cmd, capture_output=True, timeout=10, cwd=str(project_root) ) # FLY-1411: the record and the graph node name the same session. # `resolve_session_id` predicted the id from the same function # the script uses, so this reads back what it actually created # and says whether the two agree — a mismatch means something # wrote a session node between the two reads, and is worth # seeing rather than assuming away. action = {'action': 'graph_record', 'success': True} try: graph_id = json.loads(proc.stdout.decode()).get('sessionId') except (ValueError, UnicodeDecodeError): graph_id = None if graph_id and envelope is not None: matches = graph_id == f"session:{envelope['sessionId']}" action['matchesRecord'] = matches if not matches: # Only when they disagree: the id is the wrap date, and # reporting it unconditionally would make every caller's # output — the golden bridge fixtures included — change # at midnight. A mismatch is the case worth naming. action['graphSessionId'] = graph_id results['actions'].append(action) except (subprocess.TimeoutExpired, subprocess.SubprocessError, FileNotFoundError): results['actions'].append({ 'action': 'graph_record', 'success': False, 'error': 'graph_session.py failed or timed out', }) # 5. Compose result summary results['summary'] = { 'issueCount': len(issues), 'pendingCount': len(pending), 'blockerCount': len(blockers), 'filesCleanedUp': cleaned, 'summaryWritten': str(summary_file), } output_json(results) # --------------------------------------------------------------------------- # Existing commands # --------------------------------------------------------------------------- def cmd_project_update(args): body = resolve_text_input(text_arg=args.body, file_arg=args.body_file) if not body: fail("Provide body via --body, --body-file, or stdin") client = get_client() # Resolve project: explicit flag > activeProjectId > relay discovery project_id = args.project if not project_id and client.is_cloud: project_id = client.config.get("activeProjectId") if not project_id: ap = client.config.get("activeProjects", []) if ap: project_id = ap[0] result = client.project_update(args.health, body, project_id=project_id) output_json(result) def cmd_status_summary(args): client = get_client() result = client.status_summary() output_json(result) # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="FlyDocs session operations") sub = parser.add_subparsers(dest="command", required=True) # start-context sub.add_parser("start-context", help="Gather all session start data") # list-issues p = sub.add_parser("list-issues", help="Board-scoped issue fetch with fallback") # FLY-1115: was a bare 50, so a status summary silently described the # first 50 issues as though they were the whole sprint. p.add_argument("--limit", type=int, default=DEFAULT_LIST_LIMIT, help=f"Max issues to return (default {DEFAULT_LIST_LIMIT})") # wrap p = sub.add_parser("wrap", help="Perform full session wrap with cleanup") p.add_argument("--issues", nargs="*", default=[], help="Issue IDs worked on") p.add_argument("--health", choices=["onTrack", "atRisk", "offTrack"], default=None, help="Session health for project update") # FLY-1498: `--notes` is the record's `summary`, and always read like one. p.add_argument("--notes", default="", help="Session summary in prose, 2-3 sentences for a " "teammate (record `summary`, first " f"{SESSION_SUMMARY_MAX} characters)") p.add_argument("--title", default="", help="One line naming what the session did, e.g. " "\"Landed the Activity week view over live data\" " f"(max {SESSION_TITLE_MAX} characters)") p.add_argument("--pending", nargs="*", default=[], help="Pending work descriptions") p.add_argument("--blockers", nargs="*", default=[], help="Blocker descriptions") p.add_argument("--body", default=None, help="Custom project update body") # `--file` alias, catalog-wide spelling (FLY-929). p.add_argument("--body-file", "--file", default=None, dest="body_file") p.add_argument("--project", default=None, help="Target project ID") # FLY-1411: the two fields of the session record a caller can set. p.add_argument("--visibility", choices=list(VISIBILITY_CHOICES), default=None, help=f"Session record visibility (default {DEFAULT_VISIBILITY})") p.add_argument("--started-at", default=None, dest="started_at", help="Session start as ISO-8601 or unix ms " "(default: the wrap time, making the window a point)") # project-update (existing) p = sub.add_parser("project-update", help="Post a project update") p.add_argument("--health", required=True, choices=["onTrack", "atRisk", "offTrack"]) p.add_argument("--body", default=None) # `--file` alias, catalog-wide spelling (FLY-929). p.add_argument("--body-file", "--file", default=None, dest="body_file") p.add_argument("--project", default=None, help="Target project ID (defaults from activeProjects)") # status-summary (existing) sub.add_parser("status-summary", help="Show issue status counts") args = parser.parse_args() commands = { "start-context": cmd_start_context, "list-issues": cmd_list_issues, "wrap": cmd_wrap, "project-update": cmd_project_update, "status-summary": cmd_status_summary, } commands[args.command](args) if __name__ == "__main__": main()