maestro/muse_cli/commands/status.py · cgcardona/muse

1

"""muse status — show working-tree state relative to HEAD.

2

3

Output modes

4

------------

5

6

**Default (verbose, human-readable)**::

On branch main

Changes since last commit:

11

(use "muse commit -m <msg>" to record changes)

modified: beat.mid

new file: lead.mp3

deleted: scratch.mid

**--short** (condensed, one file per line)::

M beat.mid

A lead.mp3

D scratch.mid

**--porcelain** (machine-readable, stable for scripting, like git status --porcelain)::

## main

M beat.mid

A lead.mp3

D scratch.mid

**--branch** (branch and tracking info only)::

On branch main

**--sections** (group output by first directory component — musical sections)::

On branch main

## chorus

M chorus/bass.mid

A chorus/drums.mid

## verse

M verse/bass.mid

**--tracks** (group output by first directory component — instrument tracks)::

On branch main

## bass

M bass/verse.mid

A bass/chorus.mid

## drums

M drums/verse.mid

Flags are combinable where it makes sense:

57

- ``--short --sections`` → short-format codes within section groups

58

- ``--porcelain --tracks`` → porcelain codes within track groups

59

- ``--branch`` → emits only the branch line regardless of other flags

60

"""

61

from __future__ import annotations

import asyncio

import json

import logging

from collections import defaultdict

67

from pathlib import Path

68

69

import typer

70

from sqlalchemy.ext.asyncio import AsyncSession

71

72

from maestro.muse_cli._repo import require_repo

73

from maestro.muse_cli.db import get_head_snapshot_manifest, open_session

74

from maestro.muse_cli.errors import ExitCode

75

from maestro.muse_cli.merge_engine import read_merge_state

76

from maestro.muse_cli.snapshot import diff_workdir_vs_snapshot, walk_workdir

77

78

logger = logging.getLogger(__name__)

app = typer.Typer()

# ---------------------------------------------------------------------------

83

# Status code maps

84

# ---------------------------------------------------------------------------

85

86

# One-character codes for --short

87

_SHORT_CODES: dict[str, str] = {

"modified": "M",

"added": "A",

"deleted": "D",

"untracked": "?",

}

# Two-character codes for --porcelain (index + working-tree columns)

95

_PORCELAIN_CODES: dict[str, str] = {

"modified": " M",

"added": " A",

"deleted": " D",

"untracked": "??",

}

# Verbose labels for default output

103

_VERBOSE_LABELS: dict[str, str] = {

104

"modified": "modified: ",

105

"added": "new file: ",

106

"deleted": "deleted: ",

107

"untracked": "untracked: ",

}

# ---------------------------------------------------------------------------

112

# Rendering helpers

113

# ---------------------------------------------------------------------------

def _status_entries(

added: set[str],

modified: set[str],

deleted: set[str],

untracked: set[str],

) -> list[tuple[str, str]]:

122

"""Return a sorted list of (status_type, path) pairs.

123

124

Ordering: modified first, then added, deleted, untracked — mirroring

125

git's display convention of most-relevant changes first.

126

"""

127

entries: list[tuple[str, str]] = []

128

for path in sorted(modified):

129

entries.append(("modified", path))

130

for path in sorted(added):

131

entries.append(("added", path))

132

for path in sorted(deleted):

133

entries.append(("deleted", path))

134

for path in sorted(untracked):

135

entries.append(("untracked", path))

return entries

def _format_line(status: str, path: str, *, short: bool, porcelain: bool) -> str:

140

"""Format a single file line according to the active output mode.

141

142

Priority: porcelain → short → verbose.

143

144

Args:

145

status: One of ``"modified"``, ``"added"``, ``"deleted"``, ``"untracked"``.

146

path: Repo-relative path (POSIX separators).

147

short: Emit condensed ``X path`` format.

148

porcelain: Emit stable ``XY path`` format.

149

150

Returns:

151

A formatted line string (no trailing newline).

152

"""

153

if porcelain:

154

code = _PORCELAIN_CODES[status]

155

return f"{code} {path}"

156

if short:

157

code = _SHORT_CODES[status]

158

return f"{code} {path}"

159

label = _VERBOSE_LABELS[status]

160

return f"\t{label} {path}"

161

162

163

def _group_by_first_dir(entries: list[tuple[str, str]]) -> dict[str, list[tuple[str, str]]]:

164

"""Group ``(status, path)`` entries by the first directory component of *path*.

165

166

Files that live directly in the working-tree root (no sub-directory) are

167

placed under the key ``"(root)"``. This allows section/track grouping to

168

degrade gracefully when users have files at the top level.

169

"""

170

groups: dict[str, list[tuple[str, str]]] = defaultdict(list)

171

for status, path in entries:

172

slash = path.find("/")

173

key = path[:slash] if slash != -1 else "(root)"

174

groups[key].append((status, path))

return dict(groups)

def _render_flat(

entries: list[tuple[str, str]],

*,

short: bool,

porcelain: bool,

) -> None:

"""Write all entries to stdout in flat (non-grouped) order."""

185

for status, path in entries:

186

typer.echo(_format_line(status, path, short=short, porcelain=porcelain))

def _render_grouped(

entries: list[tuple[str, str]],

*,

short: bool,

porcelain: bool,

) -> None:

"""Write entries to stdout grouped under ``## <section>`` headers.

196

197

Grouping is by the first directory component of each path. Within each

198

group the entries are sorted by path. An empty line follows each group

199

to improve readability.

200

"""

201

groups = _group_by_first_dir(entries)

202

for group_name in sorted(groups.keys()):

203

typer.echo(f"## {group_name}")

204

for status, path in sorted(groups[group_name], key=lambda t: t[1]):

205

typer.echo(_format_line(status, path, short=short, porcelain=porcelain))

typer.echo("")

# ---------------------------------------------------------------------------

210

# Testable async core

211

# ---------------------------------------------------------------------------

212

213

214

async def _status_async(

215

*,

216

root: Path,

217

session: AsyncSession,

218

short: bool = False,

219

branch_only: bool = False,

220

porcelain: bool = False,

221

sections: bool = False,

222

tracks: bool = False,

223

) -> None:

224

"""Core status logic — fully injectable for tests.

225

226

Reads repo state from ``.muse/``, queries the DB session for the HEAD

227

snapshot manifest, diffs the working tree, and writes formatted output

228

via :func:`typer.echo`.

229

230

Output mode selection (evaluated in priority order):

231

232

1. ``branch_only`` → emit only the branch line and return.

233

2. ``porcelain`` → machine-readable ``XY path`` format (stable for scripts).

234

3. ``short`` → condensed ``X path`` format.

235

4. ``sections`` or ``tracks`` → group under ``## <dir>`` headers.

236

5. Default → verbose human-readable format.

237

238

``sections`` and ``tracks`` are orthogonal to ``short``/``porcelain`` and

239

can be combined with them: e.g. ``--short --sections`` emits short-format

240

lines grouped by section.

241

242

Args:

243

root: Repository root (directory containing ``.muse/``).

244

session: An open async DB session used to load the HEAD snapshot.

245

short: Emit condensed one-line-per-file output.

246

branch_only: Emit only the branch/tracking line; skip file listing.

247

porcelain: Emit machine-readable ``XY path`` format with ``## branch`` header.

248

sections: Group output by first directory component (musical sections).

249

tracks: Group output by first directory component (instrument tracks).

250

"""

251

muse_dir = root / ".muse"

252

grouped = sections or tracks

253

254

# -- Branch name --

255

head_path = muse_dir / "HEAD"

256

head_ref = head_path.read_text().strip() # "refs/heads/main"

257

branch = head_ref.rsplit("/", 1)[-1] if "/" in head_ref else head_ref

258

259

# --branch: emit only the branch line and return.

260

if branch_only:

261

typer.echo(f"On branch {branch}")

262

return

263

264

# -- In-progress merge --

265

merge_state = read_merge_state(root)

266

if merge_state is not None and merge_state.conflict_paths:

267

typer.echo(f"On branch {branch}")

268

typer.echo("")

269

typer.echo("You have unmerged paths.")

270

typer.echo(' (fix conflicts and run "muse merge --continue")')

271

typer.echo("")

272

typer.echo("Unmerged paths:")

273

for conflict_path in sorted(merge_state.conflict_paths):

274

typer.echo(f"\tboth modified: {conflict_path}")

typer.echo("")

return

# -- Check for any commits on this branch --

279

ref_path = muse_dir / head_ref

280

head_commit_id = ""

281

if ref_path.exists():

282

head_commit_id = ref_path.read_text().strip()

283

284

if not head_commit_id:

285

# No commits yet — show untracked working-tree files if any.

286

workdir = root / "muse-work"

287

untracked_files: set[str] = set()

288

if workdir.exists():

289

manifest = walk_workdir(workdir)

290

untracked_files = set(manifest.keys())

291

292

if untracked_files:

293

entries = _status_entries(set(), set(), set(), untracked_files)

294

if porcelain:

295

typer.echo(f"## {branch}")

296

_render_flat(entries, short=False, porcelain=True)

297

elif short:

298

typer.echo(f"On branch {branch}, no commits yet")

299

_render_flat(entries, short=True, porcelain=False)

300

elif grouped:

301

typer.echo(f"On branch {branch}, no commits yet")

302

typer.echo("")

303

_render_grouped(entries, short=False, porcelain=False)

304

else:

305

typer.echo(f"On branch {branch}, no commits yet")

306

typer.echo("")

307

typer.echo("Untracked files:")

308

typer.echo(' (use "muse commit -m <msg>" to record changes)')

309

typer.echo("")

310

for path in sorted(untracked_files):

311

typer.echo(f"\t{path}")

typer.echo("")

else:

if porcelain:

typer.echo(f"## {branch}")

316

else:

317

typer.echo(f"On branch {branch}, no commits yet")

318

return

319

320

# -- Load HEAD snapshot manifest from DB --

321

repo_data: dict[str, str] = json.loads((muse_dir / "repo.json").read_text())

322

repo_id = repo_data["repo_id"]

323

324

last_manifest = await get_head_snapshot_manifest(session, repo_id, branch) or {}

325

326

# -- Diff workdir vs HEAD snapshot --

327

workdir = root / "muse-work"

328

added, modified, deleted, _ = diff_workdir_vs_snapshot(workdir, last_manifest)

329

330

if not added and not modified and not deleted:

331

if porcelain:

332

typer.echo(f"## {branch}")

333

else:

334

typer.echo(f"On branch {branch}")

335

typer.echo("nothing to commit, working tree clean")

336

return

337

338

# -- Render based on active output mode --

339

entries = _status_entries(added, modified, deleted, set())

340

341

if porcelain:

342

typer.echo(f"## {branch}")

343

if grouped:

344

_render_grouped(entries, short=False, porcelain=True)

345

else:

346

_render_flat(entries, short=False, porcelain=True)

return

if short:

typer.echo(f"On branch {branch}")

351

if grouped:

352

_render_grouped(entries, short=True, porcelain=False)

353

else:

354

_render_flat(entries, short=True, porcelain=False)

return

if grouped:

typer.echo(f"On branch {branch}")

359

typer.echo("")

360

_render_grouped(entries, short=False, porcelain=False)

361

return

362

363

# -- Default verbose format --

364

typer.echo(f"On branch {branch}")

365

typer.echo("")

366

typer.echo("Changes since last commit:")

367

typer.echo(' (use "muse commit -m <msg>" to record changes)')

368

typer.echo("")

369

for path in sorted(modified):

370

typer.echo(f"\tmodified: {path}")

371

for path in sorted(added):

372

typer.echo(f"\tnew file: {path}")

373

for path in sorted(deleted):

374

typer.echo(f"\tdeleted: {path}")

typer.echo("")

# ---------------------------------------------------------------------------

379

# Typer command

380

# ---------------------------------------------------------------------------

381

382

383

@app.callback(invoke_without_command=True)

384

def status(

385

ctx: typer.Context,

386

short: bool = typer.Option(

False,

"--short",

"-s",

help="Condensed one-line-per-file output (M=modified, A=added, D=deleted, ?=untracked).",

391

),

392

branch: bool = typer.Option(

False,

"--branch",

"-b",

help="Show only the branch and tracking info line.",

397

),

398

porcelain: bool = typer.Option(

399

False,

400

"--porcelain",

401

help="Machine-readable output format (stable for scripting, like git status --porcelain).",

402

),

403

sections: bool = typer.Option(

404

False,

405

"--sections",

406

help="Group output by musical section directory (first path component under muse-work/).",

407

),

408

tracks: bool = typer.Option(

409

False,

410

"--tracks",

411

help="Group output by instrument track directory (first path component under muse-work/).",

412

),

413

) -> None:

414

"""Show the current branch and working-tree state relative to HEAD."""

415

root = require_repo()

416

417

async def _run() -> None:

418

async with open_session() as session:

await _status_async(

root=root,

session=session,

short=short,

branch_only=branch,

porcelain=porcelain,

sections=sections,

tracks=tracks,

)

try:

asyncio.run(_run())

except typer.Exit:

raise

except Exception as exc:

434

typer.echo(f"muse status failed: {exc}")

435

logger.error("muse status error: %s", exc, exc_info=True)

436

raise typer.Exit(code=ExitCode.INTERNAL_ERROR)