more organization
This commit is contained in:
+130
-163
@@ -38,16 +38,16 @@ def entry_to_str(entry: dict[str, Any]) -> str:
|
||||
Serialise a disc entry dict -> stored string.
|
||||
[C: tests/test_thinking_persistence.py:test_entry_to_str_with_thinking]
|
||||
"""
|
||||
ts = entry.get("ts", "")
|
||||
role = entry.get("role", "User")
|
||||
ts = entry.get("ts", "")
|
||||
role = entry.get("role", "User")
|
||||
content = entry.get("content", "")
|
||||
|
||||
segments = entry.get("thinking_segments")
|
||||
if segments:
|
||||
for s in segments:
|
||||
marker = s.get("marker", "thinking")
|
||||
marker = s.get("marker", "thinking")
|
||||
s_content = s.get("content", "")
|
||||
content = f"<{marker}>\n{s_content}\n</{marker}>\n{content}"
|
||||
content = f"<{marker}>\n{s_content}\n</{marker}>\n{content}"
|
||||
|
||||
if ts:
|
||||
return f"@{ts}\n{role}:\n{content}"
|
||||
@@ -64,27 +64,27 @@ def str_to_entry(raw: str, roles: list[str]) -> dict[str, Any]:
|
||||
Parse a stored string back to a disc entry dict.
|
||||
[C: tests/test_thinking_persistence.py:test_str_to_entry_with_thinking]
|
||||
"""
|
||||
ts = ""
|
||||
ts = ""
|
||||
rest = raw
|
||||
if rest.startswith("@"):
|
||||
nl = rest.find("\n")
|
||||
if nl != -1:
|
||||
ts = rest[1:nl]
|
||||
ts = rest[1:nl]
|
||||
rest = rest[nl + 1:]
|
||||
known = roles or ["User", "AI", "Vendor API", "System"]
|
||||
known = roles or ["User", "AI", "Vendor API", "System"]
|
||||
role_pat = re.compile(
|
||||
r"^(?:\[)?(" + "|".join(re.escape(r) for r in known) + r")(?:\])?:?\s*$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
parts = rest.split("\n", 1)
|
||||
parts = rest.split("\n", 1)
|
||||
matched_role = "User"
|
||||
content = rest.strip()
|
||||
content = rest.strip()
|
||||
if parts:
|
||||
m = role_pat.match(parts[0].strip())
|
||||
if m:
|
||||
raw_role = m.group(1)
|
||||
raw_role = m.group(1)
|
||||
matched_role = next((r for r in known if r.lower() == raw_role.lower()), raw_role)
|
||||
content = parts[1].strip() if len(parts) > 1 else ""
|
||||
content = parts[1].strip() if len(parts) > 1 else ""
|
||||
return {"role": matched_role, "content": content, "collapsed": False, "ts": ts}
|
||||
# ── git helpers ──────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -102,13 +102,13 @@ def get_git_commit(git_dir: str) -> str:
|
||||
|
||||
def default_discussion() -> dict[str, Any]:
|
||||
"""
|
||||
[C: tests/test_discussion_takes.py:TestDiscussionTakes.test_promote_take_renames_discussion]
|
||||
[C: tests/test_discussion_takes.py:TestDiscussionTakes.test_promote_take_renames_discussion]
|
||||
"""
|
||||
return {"git_commit": "", "last_updated": now_ts(), "history": []}
|
||||
|
||||
def default_project(name: str = "unnamed") -> dict[str, Any]:
|
||||
"""
|
||||
[C: tests/test_deepseek_infra.py:test_default_project_includes_reasoning_role, tests/test_discussion_takes.py:TestDiscussionTakes.setUp, tests/test_history_management.py:test_history_persistence_across_turns, tests/test_history_management.py:test_save_separation, tests/test_project_manager_modes.py:test_default_project_execution_mode, tests/test_project_manager_modes.py:test_load_save_execution_mode, tests/test_project_serialization.py:TestProjectSerialization.test_default_roles_include_context, tests/test_project_serialization.py:TestProjectSerialization.test_fileitem_roundtrip]
|
||||
[C: tests/test_deepseek_infra.py:test_default_project_includes_reasoning_role, tests/test_discussion_takes.py:TestDiscussionTakes.setUp, tests/test_history_management.py:test_history_persistence_across_turns, tests/test_history_management.py:test_save_separation, tests/test_project_manager_modes.py:test_default_project_execution_mode, tests/test_project_manager_modes.py:test_load_save_execution_mode, tests/test_project_serialization.py:TestProjectSerialization.test_default_roles_include_context, tests/test_project_serialization.py:TestProjectSerialization.test_fileitem_roundtrip]
|
||||
"""
|
||||
return {
|
||||
"project": {"name": name, "git_dir": "", "system_prompt": "", "execution_mode": "native"},
|
||||
@@ -120,31 +120,31 @@ def default_project(name: str = "unnamed") -> dict[str, Any]:
|
||||
"deepseek": {"reasoning_effort": "medium"},
|
||||
"agent": {
|
||||
"tools": {
|
||||
"run_powershell": True,
|
||||
"read_file": True,
|
||||
"list_directory": True,
|
||||
"search_files": True,
|
||||
"get_file_summary": True,
|
||||
"web_search": True,
|
||||
"fetch_url": True,
|
||||
"py_get_skeleton": True,
|
||||
"py_get_code_outline": True,
|
||||
"get_file_slice": True,
|
||||
"py_get_definition": True,
|
||||
"py_get_signature": True,
|
||||
"py_get_class_summary": True,
|
||||
"run_powershell": True,
|
||||
"read_file": True,
|
||||
"list_directory": True,
|
||||
"search_files": True,
|
||||
"get_file_summary": True,
|
||||
"web_search": True,
|
||||
"fetch_url": True,
|
||||
"py_get_skeleton": True,
|
||||
"py_get_code_outline": True,
|
||||
"get_file_slice": True,
|
||||
"py_get_definition": True,
|
||||
"py_get_signature": True,
|
||||
"py_get_class_summary": True,
|
||||
"py_get_var_declaration": True,
|
||||
"get_git_diff": True,
|
||||
"py_find_usages": True,
|
||||
"py_get_imports": True,
|
||||
"py_check_syntax": True,
|
||||
"py_get_hierarchy": True,
|
||||
"py_get_docstring": True,
|
||||
"get_tree": True,
|
||||
"get_ui_performance": True,
|
||||
"set_file_slice": False,
|
||||
"py_update_definition": False,
|
||||
"py_set_signature": False,
|
||||
"get_git_diff": True,
|
||||
"py_find_usages": True,
|
||||
"py_get_imports": True,
|
||||
"py_check_syntax": True,
|
||||
"py_get_hierarchy": True,
|
||||
"py_get_docstring": True,
|
||||
"get_tree": True,
|
||||
"get_ui_performance": True,
|
||||
"set_file_slice": False,
|
||||
"py_update_definition": False,
|
||||
"py_set_signature": False,
|
||||
"py_set_var_declaration": False,
|
||||
}
|
||||
},
|
||||
@@ -163,23 +163,19 @@ def default_project(name: str = "unnamed") -> dict[str, Any]:
|
||||
|
||||
def get_history_path(project_path: Union[str, Path]) -> Path:
|
||||
"""
|
||||
|
||||
Return the Path to the sibling history TOML file for a given project.
|
||||
[C: tests/test_history_management.py:test_save_separation]
|
||||
Return the Path to the sibling history TOML file for a given project.
|
||||
[C: tests/test_history_management.py:test_save_separation]
|
||||
"""
|
||||
p = Path(project_path)
|
||||
return p.parent / f"{p.stem}_history.toml"
|
||||
|
||||
def load_project(path: Union[str, Path]) -> dict[str, Any]:
|
||||
"""
|
||||
|
||||
|
||||
Load a project TOML file.
|
||||
Automatically migrates legacy 'discussion' keys to a sibling history file.
|
||||
[C: tests/test_history_management.py:test_history_persistence_across_turns, tests/test_history_management.py:test_migration_on_load, tests/test_project_manager_modes.py:test_load_save_execution_mode, tests/test_project_serialization.py:TestProjectSerialization.test_backward_compatibility_strings, tests/test_project_serialization.py:TestProjectSerialization.test_fileitem_roundtrip]
|
||||
Load a project TOML file.
|
||||
Automatically migrates legacy 'discussion' keys to a sibling history file.
|
||||
[C: tests/test_history_management.py:test_history_persistence_across_turns, tests/test_history_management.py:test_migration_on_load, tests/test_project_manager_modes.py:test_load_save_execution_mode, tests/test_project_serialization.py:TestProjectSerialization.test_backward_compatibility_strings, tests/test_project_serialization.py:TestProjectSerialization.test_fileitem_roundtrip]
|
||||
"""
|
||||
with open(path, "rb") as f:
|
||||
proj = tomllib.load(f)
|
||||
with open(path, "rb") as f: proj = tomllib.load(f)
|
||||
# Deserialise FileItems in files.paths
|
||||
if "files" in proj and "paths" in proj["files"]:
|
||||
from src import models
|
||||
@@ -198,9 +194,8 @@ def load_project(path: Union[str, Path]) -> dict[str, Any]:
|
||||
|
||||
def load_history(project_path: Union[str, Path]) -> dict[str, Any]:
|
||||
"""
|
||||
|
||||
Load the segregated discussion history from its dedicated TOML file.
|
||||
[C: tests/test_thinking_persistence.py:test_save_and_load_history_with_thinking_segments]
|
||||
Load the segregated discussion history from its dedicated TOML file.
|
||||
[C: tests/test_thinking_persistence.py:test_save_and_load_history_with_thinking_segments]
|
||||
"""
|
||||
hist_path = get_history_path(project_path)
|
||||
if hist_path.exists():
|
||||
@@ -210,31 +205,25 @@ def load_history(project_path: Union[str, Path]) -> dict[str, Any]:
|
||||
|
||||
def clean_nones(data: Any) -> Any:
|
||||
"""
|
||||
|
||||
Recursively remove None values from a dictionary/list.
|
||||
[C: tests/test_thinking_persistence.py:test_clean_nones_removes_thinking]
|
||||
Recursively remove None values from a dictionary/list.
|
||||
[C: tests/test_thinking_persistence.py:test_clean_nones_removes_thinking]
|
||||
"""
|
||||
if isinstance(data, dict):
|
||||
return {k: clean_nones(v) for k, v in data.items() if v is not None}
|
||||
elif isinstance(data, list):
|
||||
return [clean_nones(v) for v in data if v is not None]
|
||||
if isinstance(data, dict): return {k: clean_nones(v) for k, v in data.items() if v is not None}
|
||||
elif isinstance(data, list): return [clean_nones(v) for v in data if v is not None]
|
||||
return data
|
||||
|
||||
def save_project(proj: dict[str, Any], path: Union[str, Path], disc_data: Optional[dict[str, Any]] = None) -> None:
|
||||
"""
|
||||
|
||||
|
||||
Save the project TOML.
|
||||
If 'discussion' is present in proj, it is moved to the sibling history file.
|
||||
[C: tests/test_history_management.py:test_history_persistence_across_turns, tests/test_history_management.py:test_save_separation, tests/test_project_manager_modes.py:test_load_save_execution_mode, tests/test_project_serialization.py:TestProjectSerialization.test_fileitem_roundtrip, tests/test_thinking_persistence.py:test_save_and_load_history_with_thinking_segments]
|
||||
Save the project TOML.
|
||||
If 'discussion' is present in proj, it is moved to the sibling history file.
|
||||
[C: tests/test_history_management.py:test_history_persistence_across_turns, tests/test_history_management.py:test_save_separation, tests/test_project_manager_modes.py:test_load_save_execution_mode, tests/test_project_serialization.py:TestProjectSerialization.test_fileitem_roundtrip, tests/test_thinking_persistence.py:test_save_and_load_history_with_thinking_segments]
|
||||
"""
|
||||
proj = clean_nones(proj)
|
||||
# Serialise FileItems
|
||||
if "files" in proj and "paths" in proj["files"]:
|
||||
proj["files"]["paths"] = [p.to_dict() if hasattr(p, "to_dict") else p for p in proj["files"]["paths"]]
|
||||
if "discussion" in proj:
|
||||
if disc_data is None:
|
||||
disc_data = proj["discussion"]
|
||||
if disc_data is None: disc_data = proj["discussion"]
|
||||
proj = dict(proj)
|
||||
del proj["discussion"]
|
||||
proj = clean_nones(proj)
|
||||
@@ -252,13 +241,12 @@ def migrate_from_legacy_config(cfg: dict[str, Any]) -> dict[str, Any]:
|
||||
name = cfg.get("output", {}).get("namespace", "project")
|
||||
proj = default_project(name)
|
||||
for key in ("output", "files", "screenshots"):
|
||||
if key in cfg:
|
||||
proj[key] = dict(cfg[key])
|
||||
disc = cfg.get("discussion", {})
|
||||
if key in cfg: proj[key] = dict(cfg[key])
|
||||
disc = cfg.get("discussion", {})
|
||||
proj["discussion"]["roles"] = disc.get("roles", ["User", "AI", "Vendor API", "System", "Context"])
|
||||
main_disc = proj["discussion"]["discussions"]["main"]
|
||||
main_disc["history"] = disc.get("history", [])
|
||||
main_disc["last_updated"] = now_ts()
|
||||
main_disc = proj["discussion"]["discussions"]["main"]
|
||||
main_disc["history"] = disc.get("history", [])
|
||||
main_disc["last_updated"] = now_ts()
|
||||
return proj
|
||||
# ── flat config for aggregate.run() ─────────────────────────────────────────
|
||||
|
||||
@@ -268,9 +256,9 @@ def flat_config(proj: dict[str, Any], disc_name: Optional[str] = None, track_id:
|
||||
if track_id:
|
||||
history = load_track_history(track_id, proj.get("files", {}).get("base_dir", "."))
|
||||
else:
|
||||
name = disc_name or disc_sec.get("active", "main")
|
||||
name = disc_name or disc_sec.get("active", "main")
|
||||
disc_data = disc_sec.get("discussions", {}).get(name, {})
|
||||
history = disc_data.get("history", [])
|
||||
history = disc_data.get("history", [])
|
||||
return {
|
||||
"project": proj.get("project", {}),
|
||||
"output": proj.get("output", {}),
|
||||
@@ -286,187 +274,169 @@ def flat_config(proj: dict[str, Any], disc_name: Optional[str] = None, track_id:
|
||||
|
||||
def save_track_state(track_id: str, state: 'TrackState', base_dir: Union[str, Path] = ".") -> None:
|
||||
"""
|
||||
|
||||
|
||||
Saves a TrackState object to conductor/tracks/<track_id>/state.toml.
|
||||
[C: tests/test_project_manager_tracks.py:test_get_all_tracks_with_state, tests/test_track_state_persistence.py:test_track_state_persistence]
|
||||
Saves a TrackState object to conductor/tracks/<track_id>/state.toml.
|
||||
[C: tests/test_project_manager_tracks.py:test_get_all_tracks_with_state, tests/test_track_state_persistence.py:test_track_state_persistence]
|
||||
"""
|
||||
track_dir = paths.get_track_state_dir(track_id, project_path=str(base_dir))
|
||||
track_dir.mkdir(parents=True, exist_ok=True)
|
||||
state_file = track_dir / "state.toml"
|
||||
data = clean_nones(state.to_dict())
|
||||
with open(state_file, "wb") as f:
|
||||
tomli_w.dump(data, f)
|
||||
with open(state_file, "wb") as f: tomli_w.dump(data, f)
|
||||
|
||||
def load_track_state(track_id: str, base_dir: Union[str, Path] = ".") -> Optional['TrackState']:
|
||||
"""
|
||||
|
||||
|
||||
Loads a TrackState object from conductor/tracks/<track_id>/state.toml.
|
||||
[C: tests/test_track_state_persistence.py:test_track_state_persistence]
|
||||
Loads a TrackState object from conductor/tracks/<track_id>/state.toml.
|
||||
[C: tests/test_track_state_persistence.py:test_track_state_persistence]
|
||||
"""
|
||||
from src.models import TrackState
|
||||
state_file = paths.get_track_state_dir(track_id, project_path=str(base_dir)) / 'state.toml'
|
||||
if not state_file.exists():
|
||||
return None
|
||||
with open(state_file, "rb") as f:
|
||||
data = tomllib.load(f)
|
||||
if not state_file.exists(): return None
|
||||
with open(state_file, "rb") as f: data = tomllib.load(f)
|
||||
return TrackState.from_dict(data)
|
||||
|
||||
def load_track_history(track_id: str, base_dir: Union[str, Path] = ".") -> list[str]:
|
||||
"""
|
||||
|
||||
|
||||
Loads the discussion history for a specific track from its state.toml.
|
||||
Returns a list of entry strings formatted with @timestamp.
|
||||
Loads the discussion history for a specific track from its state.toml.
|
||||
Returns a list of entry strings formatted with @timestamp.
|
||||
"""
|
||||
state = load_track_state(track_id, base_dir)
|
||||
if not state:
|
||||
return []
|
||||
if not state: return []
|
||||
history: list[str] = []
|
||||
for entry in state.discussion:
|
||||
e = dict(entry)
|
||||
e = dict(entry)
|
||||
ts = e.get("ts")
|
||||
if isinstance(ts, datetime.datetime):
|
||||
e["ts"] = ts.strftime(TS_FMT)
|
||||
if isinstance(ts, datetime.datetime): e["ts"] = ts.strftime(TS_FMT)
|
||||
history.append(entry_to_str(e))
|
||||
return history
|
||||
|
||||
def save_track_history(track_id: str, history: list[str], base_dir: Union[str, Path] = ".") -> None:
|
||||
"""
|
||||
|
||||
|
||||
Saves the discussion history for a specific track to its state.toml.
|
||||
'history' is expected to be a list of formatted strings.
|
||||
Saves the discussion history for a specific track to its state.toml.
|
||||
'history' is expected to be a list of formatted strings.
|
||||
"""
|
||||
state = load_track_state(track_id, base_dir)
|
||||
if not state:
|
||||
return
|
||||
roles = ["User", "AI", "Vendor API", "System", "Reasoning"]
|
||||
roles = ["User", "AI", "Vendor API", "System", "Reasoning"]
|
||||
entries = [str_to_entry(h, roles) for h in history]
|
||||
state.discussion = entries
|
||||
save_track_state(track_id, state, base_dir)
|
||||
|
||||
def get_all_tracks(base_dir: Union[str, Path] = ".") -> list[dict[str, Any]]:
|
||||
"""
|
||||
|
||||
|
||||
Scans the conductor/tracks/ directory and returns a list of dictionaries
|
||||
containing track metadata: 'id', 'title', 'status', 'complete', 'total',
|
||||
and 'progress' (0.0 to 1.0).
|
||||
Handles missing or malformed metadata.json or state.toml by falling back
|
||||
to available info or defaults.
|
||||
[C: tests/test_project_manager_tracks.py:test_get_all_tracks_empty, tests/test_project_manager_tracks.py:test_get_all_tracks_malformed, tests/test_project_manager_tracks.py:test_get_all_tracks_with_metadata_json, tests/test_project_manager_tracks.py:test_get_all_tracks_with_state, tests/test_project_paths.py:test_get_all_tracks_project_specific]
|
||||
Scans the conductor/tracks/ directory and returns a list of dictionaries
|
||||
containing track metadata: 'id', 'title', 'status', 'complete', 'total',
|
||||
and 'progress' (0.0 to 1.0).
|
||||
Handles missing or malformed metadata.json or state.toml by falling back
|
||||
to available info or defaults.
|
||||
[C: tests/test_project_manager_tracks.py:test_get_all_tracks_empty, tests/test_project_manager_tracks.py:test_get_all_tracks_malformed, tests/test_project_manager_tracks.py:test_get_all_tracks_with_metadata_json, tests/test_project_manager_tracks.py:test_get_all_tracks_with_state, tests/test_project_paths.py:test_get_all_tracks_project_specific]
|
||||
"""
|
||||
tracks_dir = paths.get_tracks_dir(project_path=str(base_dir))
|
||||
if not tracks_dir.exists():
|
||||
return []
|
||||
if not tracks_dir.exists(): return []
|
||||
|
||||
results: list[dict[str, Any]] = []
|
||||
for entry in tracks_dir.iterdir():
|
||||
if not entry.is_dir():
|
||||
continue
|
||||
if not entry.is_dir(): continue
|
||||
|
||||
track_id = entry.name
|
||||
track_info: dict[str, Any] = {
|
||||
"id": track_id,
|
||||
"title": track_id,
|
||||
"status": "unknown",
|
||||
"id": track_id,
|
||||
"title": track_id,
|
||||
"status": "unknown",
|
||||
"complete": 0,
|
||||
"total": 0,
|
||||
"total": 0,
|
||||
"progress": 0.0
|
||||
}
|
||||
state_found = False
|
||||
|
||||
try:
|
||||
state = load_track_state(track_id, base_dir)
|
||||
if state:
|
||||
track_info["id"] = state.metadata.id or track_id
|
||||
track_info["title"] = state.metadata.name or track_id
|
||||
track_info["id"] = state.metadata.id or track_id
|
||||
track_info["title"] = state.metadata.name or track_id
|
||||
track_info["status"] = state.metadata.status or "unknown"
|
||||
progress = calculate_track_progress(state.tasks)
|
||||
track_info["complete"] = progress["completed"]
|
||||
track_info["total"] = progress["total"]
|
||||
track_info["total"] = progress["total"]
|
||||
track_info["progress"] = progress["percentage"] / 100.0
|
||||
state_found = True
|
||||
state_found = True
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not state_found:
|
||||
metadata_file = entry / "metadata.json"
|
||||
if metadata_file.exists():
|
||||
try:
|
||||
with open(metadata_file, "r") as f:
|
||||
data = json.load(f)
|
||||
track_info["id"] = data.get("id", data.get("track_id", track_id))
|
||||
track_info["title"] = data.get("title", data.get("name", data.get("description", track_id)))
|
||||
data = json.load(f)
|
||||
track_info["id"] = data.get("id", data.get("track_id", track_id))
|
||||
track_info["title"] = data.get("title", data.get("name", data.get("description", track_id)))
|
||||
track_info["status"] = data.get("status", "unknown")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if track_info["total"] == 0:
|
||||
plan_file = entry / "plan.md"
|
||||
if plan_file.exists():
|
||||
try:
|
||||
with open(plan_file, "r", encoding="utf-8") as f:
|
||||
content = f.read()
|
||||
tasks = re.findall(r"^[ \t]*- \[[ x~]\] .*", content, re.MULTILINE)
|
||||
content = f.read()
|
||||
tasks = re.findall(r"^[ \t]*- \[[ x~]\] .*", content, re.MULTILINE)
|
||||
completed_tasks = re.findall(r"^[ \t]*- \[x\] .*", content, re.MULTILINE)
|
||||
track_info["total"] = len(tasks)
|
||||
track_info["total"] = len(tasks)
|
||||
track_info["complete"] = len(completed_tasks)
|
||||
if track_info["total"] > 0:
|
||||
track_info["progress"] = float(track_info["complete"]) / track_info["total"]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
results.append(track_info)
|
||||
return results
|
||||
|
||||
def calculate_track_progress(tickets: list) -> dict:
|
||||
"""
|
||||
|
||||
|
||||
Calculates track progress based on ticket statuses.
|
||||
percentage (float), completed (int), total (int), in_progress (int), blocked (int), todo (int)
|
||||
[C: tests/test_progress_viz.py:test_calculate_track_progress_all_completed, tests/test_progress_viz.py:test_calculate_track_progress_all_todo, tests/test_progress_viz.py:test_calculate_track_progress_empty, tests/test_progress_viz.py:test_calculate_track_progress_mixed]
|
||||
Calculates track progress based on ticket statuses.
|
||||
percentage (float), completed (int), total (int), in_progress (int), blocked (int), todo (int)
|
||||
[C: tests/test_progress_viz.py:test_calculate_track_progress_all_completed, tests/test_progress_viz.py:test_calculate_track_progress_all_todo, tests/test_progress_viz.py:test_calculate_track_progress_empty, tests/test_progress_viz.py:test_calculate_track_progress_mixed]
|
||||
"""
|
||||
total = len(tickets)
|
||||
if total == 0:
|
||||
return {
|
||||
"percentage": 0.0,
|
||||
"completed": 0,
|
||||
"total": 0,
|
||||
"percentage": 0.0,
|
||||
"completed": 0,
|
||||
"total": 0,
|
||||
"in_progress": 0,
|
||||
"blocked": 0,
|
||||
"todo": 0
|
||||
"blocked": 0,
|
||||
"todo": 0
|
||||
}
|
||||
|
||||
completed = sum(1 for t in tickets if t.status == "completed")
|
||||
completed = sum(1 for t in tickets if t.status == "completed")
|
||||
in_progress = sum(1 for t in tickets if t.status == "in_progress")
|
||||
blocked = sum(1 for t in tickets if t.status == "blocked")
|
||||
todo = sum(1 for t in tickets if t.status == "todo")
|
||||
|
||||
percentage = (completed / total) * 100.0
|
||||
blocked = sum(1 for t in tickets if t.status == "blocked")
|
||||
todo = sum(1 for t in tickets if t.status == "todo")
|
||||
percentage = (completed / total) * 100.0
|
||||
|
||||
return {
|
||||
"percentage": float(percentage),
|
||||
"completed": completed,
|
||||
"total": total,
|
||||
"percentage": float(percentage),
|
||||
"completed": completed,
|
||||
"total": total,
|
||||
"in_progress": in_progress,
|
||||
"blocked": blocked,
|
||||
"todo": todo
|
||||
"blocked": blocked,
|
||||
"todo": todo
|
||||
}
|
||||
|
||||
|
||||
def branch_discussion(project_dict: dict, source_id: str, new_id: str, message_index: int) -> None:
|
||||
"""
|
||||
|
||||
|
||||
Creates a new discussion in project_dict['discussion']['discussions'] by copying
|
||||
the history from source_id up to (and including) message_index, and sets active to new_id.
|
||||
[C: tests/test_discussion_takes.py:TestDiscussionTakes.test_branch_discussion_creates_new_take]
|
||||
Creates a new discussion in project_dict['discussion']['discussions'] by copying
|
||||
the history from source_id up to (and including) message_index, and sets active to new_id.
|
||||
[C: tests/test_discussion_takes.py:TestDiscussionTakes.test_branch_discussion_creates_new_take]
|
||||
"""
|
||||
if "discussion" not in project_dict or "discussions" not in project_dict["discussion"]:
|
||||
return
|
||||
if source_id not in project_dict["discussion"]["discussions"]:
|
||||
return
|
||||
if "discussion" not in project_dict or "discussions" not in project_dict["discussion"]: return
|
||||
if source_id not in project_dict["discussion"]["discussions"]: return
|
||||
|
||||
source_disc = project_dict["discussion"]["discussions"][source_id]
|
||||
new_disc = default_discussion()
|
||||
new_disc = default_discussion()
|
||||
new_disc["git_commit"] = source_disc.get("git_commit", "")
|
||||
# Copy history up to and including message_index
|
||||
new_disc["history"] = source_disc["history"][:message_index + 1]
|
||||
@@ -476,14 +446,11 @@ def branch_discussion(project_dict: dict, source_id: str, new_id: str, message_i
|
||||
|
||||
def promote_take(project_dict: dict, take_id: str, new_id: str) -> None:
|
||||
"""
|
||||
|
||||
Renames a take_id to new_id in the discussions dict.
|
||||
[C: tests/test_discussion_takes.py:TestDiscussionTakes.test_promote_take_renames_discussion]
|
||||
Renames a take_id to new_id in the discussions dict.
|
||||
[C: tests/test_discussion_takes.py:TestDiscussionTakes.test_promote_take_renames_discussion]
|
||||
"""
|
||||
if "discussion" not in project_dict or "discussions" not in project_dict["discussion"]:
|
||||
return
|
||||
if take_id not in project_dict["discussion"]["discussions"]:
|
||||
return
|
||||
if "discussion" not in project_dict or "discussions" not in project_dict["discussion"]: return
|
||||
if take_id not in project_dict["discussion"]["discussions"]: return
|
||||
|
||||
disc = project_dict["discussion"]["discussions"].pop(take_id)
|
||||
project_dict["discussion"]["discussions"][new_id] = disc
|
||||
|
||||
Reference in New Issue
Block a user