PlanOpticon

feat(skills): add 9 planning skills and wire up agent CLI Skills: project_plan, prd, roadmap, task_breakdown, doc_generator, requirements_chat, github_issues, cli_adapter, artifact_export - Auto-import all skills on agent startup - Export uses structured directory layout with manifest.json - GitHub integration shells out to gh CLI when available - CLI adapter supports github, jira, linear with dry-run mode

lmata 2026-03-07 22:05 trunk
Commit 26bc92c6a4c3a574793a3db9cfa953b8092f0e8c8b47f351dfe29d6a5029ab52
Parent 18eaec6671c407a…
11 files changed +12 +94 +99 +76 +93 +70 +74 +95 +68 +76 +7 -7
~ video_processor/agent/skills/__init__.py + video_processor/agent/skills/artifact_export.py + video_processor/agent/skills/cli_adapter.py + video_processor/agent/skills/doc_generator.py + video_processor/agent/skills/github_integration.py + video_processor/agent/skills/prd.py + video_processor/agent/skills/project_plan.py + video_processor/agent/skills/requirements_chat.py + video_processor/agent/skills/roadmap.py + video_processor/agent/skills/task_breakdown.py ~ video_processor/cli/commands.py
M video_processor/agent/skills/__init__.py
+12
--- video_processor/agent/skills/__init__.py
+++ video_processor/agent/skills/__init__.py
@@ -1,7 +1,19 @@
11
 """Agent skill system for PlanOpticon."""
22
3
+# Import skill modules so they self-register via register_skill().
4
+from video_processor.agent.skills import ( # noqa: F401
5
+ artifact_export,
6
+ cli_adapter,
7
+ doc_generator,
8
+ github_integration,
9
+ prd,
10
+ project_plan,
11
+ requirements_chat,
12
+ roadmap,
13
+ task_breakdown,
14
+)
315
 from video_processor.agent.skills.base import (
416
 AgentContext,
517
 Artifact,
618
 Skill,
719
 get_skill,
820
921
ADDED video_processor/agent/skills/artifact_export.py
1022
ADDED video_processor/agent/skills/cli_adapter.py
1123
ADDED video_processor/agent/skills/doc_generator.py
1224
ADDED video_processor/agent/skills/github_integration.py
1325
ADDED video_processor/agent/skills/prd.py
1426
ADDED video_processor/agent/skills/project_plan.py
1527
ADDED video_processor/agent/skills/requirements_chat.py
1628
ADDED video_processor/agent/skills/roadmap.py
1729
ADDED video_processor/agent/skills/task_breakdown.py
--- video_processor/agent/skills/__init__.py
+++ video_processor/agent/skills/__init__.py
@@ -1,7 +1,19 @@
1 """Agent skill system for PlanOpticon."""
2
 
 
 
 
 
 
 
 
 
 
 
 
3 from video_processor.agent.skills.base import (
4 AgentContext,
5 Artifact,
6 Skill,
7 get_skill,
8
9 DDED video_processor/agent/skills/artifact_export.py
10 DDED video_processor/agent/skills/cli_adapter.py
11 DDED video_processor/agent/skills/doc_generator.py
12 DDED video_processor/agent/skills/github_integration.py
13 DDED video_processor/agent/skills/prd.py
14 DDED video_processor/agent/skills/project_plan.py
15 DDED video_processor/agent/skills/requirements_chat.py
16 DDED video_processor/agent/skills/roadmap.py
17 DDED video_processor/agent/skills/task_breakdown.py
--- video_processor/agent/skills/__init__.py
+++ video_processor/agent/skills/__init__.py
@@ -1,7 +1,19 @@
1 """Agent skill system for PlanOpticon."""
2
3 # Import skill modules so they self-register via register_skill().
4 from video_processor.agent.skills import ( # noqa: F401
5 artifact_export,
6 cli_adapter,
7 doc_generator,
8 github_integration,
9 prd,
10 project_plan,
11 requirements_chat,
12 roadmap,
13 task_breakdown,
14 )
15 from video_processor.agent.skills.base import (
16 AgentContext,
17 Artifact,
18 Skill,
19 get_skill,
20
21 DDED video_processor/agent/skills/artifact_export.py
22 DDED video_processor/agent/skills/cli_adapter.py
23 DDED video_processor/agent/skills/doc_generator.py
24 DDED video_processor/agent/skills/github_integration.py
25 DDED video_processor/agent/skills/prd.py
26 DDED video_processor/agent/skills/project_plan.py
27 DDED video_processor/agent/skills/requirements_chat.py
28 DDED video_processor/agent/skills/roadmap.py
29 DDED video_processor/agent/skills/task_breakdown.py
A video_processor/agent/skills/artifact_export.py
+94
--- a/video_processor/agent/skills/artifact_export.py
+++ b/video_processor/agent/skills/artifact_export.py
@@ -0,0 +1,94 @@
1
+"""Skill: Export artifacts in agent-ready formats to a directory structure."""
2
+
3
+import json
4
+from pathlib import Path
5
+
6
+from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill
7
+
8
+# Maps artifact_type to output filename
9
+_TYPE_TO_FILE = {
10
+ "project_plan": "project_plan.md",
11
+ "prd": "prd.md",
12
+ "roadmap": "roadmap.md",
13
+ "task_list": "tasks.json",
14
+ "issues": "issues.json",
15
+ "requirements": "requirements.json",
16
+}
17
+
18
+
19
+def _write_artifact(artifact: Artifact, output_dir: Path) -> dict:
20
+ """Write a single artifact to the appropriate file. Returns manifest entry."""
21
+ filename = _TYPE_TO_FILE.get(artifact.artifact_type)
22
+ if filename:
23
+ dest = output_dir / filename
24
+ elif artifact.artifact_type == "document":
25
+ docs_dir = output_dir / "docs"
26
+ docs_dir.mkdir(parents=True, exist_ok=True)
27
+ safe_name = artifact.name.replace(" ", "_").replace("/", "_").lower()
28
+ ext = ".json" if artifact.format == "json" else ".md"
29
+ dest = docs_dir / f"{safe_name}{ext}"
30
+ else:
31
+ safe_name = artifact.name.replace(" ", "_").replace("/", "_").lower()
32
+ ext = ".json" if artifact.format == "json" else ".md"
33
+ dest = output_dir / f"{safe_name}{ext}"
34
+
35
+ dest.write_text(artifact.content, encoding="utf-8")
36
+ return {
37
+ "file": str(dest),
38
+ "name": artifact.name,
39
+ "artifact_type": artifact.artifact_type,
40
+ "format": artifact.format,
41
+ }
42
+
43
+
44
+class ArtifactExportSkill(Skill):
45
+ name = "artifact_export"
46
+ description = "Export artifacts in agent-ready formats"
47
+
48
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
49
+ output_dir = Path(kwargs.get("output_dir", "plan"))
50
+ output_dir.mkdir(parents=True, exist_ok=True)
51
+
52
+ manifest_entries = []
53
+ for artifact in context.artifacts:
54
+ entry = _write_artifact(artifact, output_dir)
55
+ manifest_entries.append(entry)
56
+
57
+ manifest = {
58
+ "artifact_count": len(manifest_entries),
59
+ "output_dir": str(output_dir),
60
+ "files": manifest_entries,
61
+ }
62
+ manifest_path = output_dir / "manifest.json"
63
+ manifest_json = json.dumps(manifest, indent=2)
64
+ manifest_path.write_text(manifest_json, encoding="utf-8")
65
+
66
+ return Artifact(
67
+ name="Export Manifest",
68
+ content=manifest_json,
69
+ artifact_type="export_manifest",
70
+ format="json",
71
+ )
72
+
73
+
74
+register_skill(ArtifactExportSkill())
75
+
76
+
77
+def export_artifacts(artifacts: list, output_dir: Path) -> dict:
78
+ """Standalone helper: export a list of Artifact objects to a directory."""
79
+ output_dir = Path(output_dir)
80
+ output_dir.mkdir(parents=True, exist_ok=True)
81
+
82
+ manifest_entries = []
83
+ for artifact in artifacts:
84
+ entry = _write_artifact(artifact, output_dir)
85
+ manifest_entries.append(entry)
86
+
87
+ manifest = {
88
+ "artifact_count": len(manifest_entries),
89
+ "output_dir": str(output_dir),
90
+ "files": manifest_entries,
91
+ }
92
+ manifest_path = output_dir / "manifest.json"
93
+ manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
94
+ return manifest
--- a/video_processor/agent/skills/artifact_export.py
+++ b/video_processor/agent/skills/artifact_export.py
@@ -0,0 +1,94 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/artifact_export.py
+++ b/video_processor/agent/skills/artifact_export.py
@@ -0,0 +1,94 @@
1 """Skill: Export artifacts in agent-ready formats to a directory structure."""
2
3 import json
4 from pathlib import Path
5
6 from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill
7
8 # Maps artifact_type to output filename
9 _TYPE_TO_FILE = {
10 "project_plan": "project_plan.md",
11 "prd": "prd.md",
12 "roadmap": "roadmap.md",
13 "task_list": "tasks.json",
14 "issues": "issues.json",
15 "requirements": "requirements.json",
16 }
17
18
19 def _write_artifact(artifact: Artifact, output_dir: Path) -> dict:
20 """Write a single artifact to the appropriate file. Returns manifest entry."""
21 filename = _TYPE_TO_FILE.get(artifact.artifact_type)
22 if filename:
23 dest = output_dir / filename
24 elif artifact.artifact_type == "document":
25 docs_dir = output_dir / "docs"
26 docs_dir.mkdir(parents=True, exist_ok=True)
27 safe_name = artifact.name.replace(" ", "_").replace("/", "_").lower()
28 ext = ".json" if artifact.format == "json" else ".md"
29 dest = docs_dir / f"{safe_name}{ext}"
30 else:
31 safe_name = artifact.name.replace(" ", "_").replace("/", "_").lower()
32 ext = ".json" if artifact.format == "json" else ".md"
33 dest = output_dir / f"{safe_name}{ext}"
34
35 dest.write_text(artifact.content, encoding="utf-8")
36 return {
37 "file": str(dest),
38 "name": artifact.name,
39 "artifact_type": artifact.artifact_type,
40 "format": artifact.format,
41 }
42
43
44 class ArtifactExportSkill(Skill):
45 name = "artifact_export"
46 description = "Export artifacts in agent-ready formats"
47
48 def execute(self, context: AgentContext, **kwargs) -> Artifact:
49 output_dir = Path(kwargs.get("output_dir", "plan"))
50 output_dir.mkdir(parents=True, exist_ok=True)
51
52 manifest_entries = []
53 for artifact in context.artifacts:
54 entry = _write_artifact(artifact, output_dir)
55 manifest_entries.append(entry)
56
57 manifest = {
58 "artifact_count": len(manifest_entries),
59 "output_dir": str(output_dir),
60 "files": manifest_entries,
61 }
62 manifest_path = output_dir / "manifest.json"
63 manifest_json = json.dumps(manifest, indent=2)
64 manifest_path.write_text(manifest_json, encoding="utf-8")
65
66 return Artifact(
67 name="Export Manifest",
68 content=manifest_json,
69 artifact_type="export_manifest",
70 format="json",
71 )
72
73
74 register_skill(ArtifactExportSkill())
75
76
77 def export_artifacts(artifacts: list, output_dir: Path) -> dict:
78 """Standalone helper: export a list of Artifact objects to a directory."""
79 output_dir = Path(output_dir)
80 output_dir.mkdir(parents=True, exist_ok=True)
81
82 manifest_entries = []
83 for artifact in artifacts:
84 entry = _write_artifact(artifact, output_dir)
85 manifest_entries.append(entry)
86
87 manifest = {
88 "artifact_count": len(manifest_entries),
89 "output_dir": str(output_dir),
90 "files": manifest_entries,
91 }
92 manifest_path = output_dir / "manifest.json"
93 manifest_path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
94 return manifest
A video_processor/agent/skills/cli_adapter.py
+99
--- a/video_processor/agent/skills/cli_adapter.py
+++ b/video_processor/agent/skills/cli_adapter.py
@@ -0,0 +1,99 @@
1
+"""Skill: Push artifacts to external tools via their CLIs."""
2
+
3
+import json
4
+import shutil
5
+import subprocess
6
+from typing import List
7
+
8
+from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill
9
+
10
+
11
+def _format_github(artifact: Artifact) -> List[str]:
12
+ """Convert artifact to gh CLI commands."""
13
+ items = json.loads(artifact.content) if artifact.format == "json" else []
14
+ cmds = []
15
+ for item in items:
16
+ cmd = f"gh issue create --title {json.dumps(item.get('title', ''))}"
17
+ if item.get("body"):
18
+ cmd += f" --body {json.dumps(item['body'])}"
19
+ for label in item.get("labels", []):
20
+ cmd += f" --label {json.dumps(label)}"
21
+ cmds.append(cmd)
22
+ return cmds
23
+
24
+
25
+def _format_jira(artifact: Artifact) -> List[str]:
26
+ """Convert artifact to jira-cli commands."""
27
+ items = json.loads(artifact.content) if artifact.format == "json" else []
28
+ return [
29
+ f"jira issue create --summary {json.dumps(item.get('title', ''))}"
30
+ f" --description {json.dumps(item.get('body', item.get('description', '')))}"
31
+ for item in items
32
+ ]
33
+
34
+
35
+def _format_linear(artifact: Artifact) -> List[str]:
36
+ """Convert artifact to linear CLI commands."""
37
+ items = json.loads(artifact.content) if artifact.format == "json" else []
38
+ return [
39
+ f"linear issue create --title {json.dumps(item.get('title', ''))}"
40
+ f" --description {json.dumps(item.get('body', item.get('description', '')))}"
41
+ for item in items
42
+ ]
43
+
44
+
45
+_adapters = {"github": _format_github, "jira": _format_jira, "linear": _format_linear}
46
+
47
+
48
+def run_commands(commands: List[str], dry_run: bool = True) -> List[dict]:
49
+ """Execute CLI commands. In dry_run mode, just return what would run."""
50
+ results = []
51
+ for cmd in commands:
52
+ if dry_run:
53
+ results.append({"command": cmd, "status": "dry_run"})
54
+ else:
55
+ proc = subprocess.run(cmd, shell=True, capture_output=True, text=True)
56
+ results.append(
57
+ {
58
+ "command": cmd,
59
+ "returncode": proc.returncode,
60
+ "stdout": proc.stdout.strip(),
61
+ "stderr": proc.stderr.strip(),
62
+ }
63
+ )
64
+ return results
65
+
66
+
67
+class CLIAdapterSkill(Skill):
68
+ name = "cli_adapter"
69
+ description = "Push artifacts to external tools via their CLIs"
70
+
71
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
72
+ tool = kwargs.get("tool", "github")
73
+ artifact = kwargs.get("artifact")
74
+ if artifact is None and context.artifacts:
75
+ artifact = context.artifacts[-1]
76
+ if artifact is None:
77
+ return Artifact(
78
+ name="CLI Commands", content="[]", artifact_type="cli_commands", format="json"
79
+ )
80
+
81
+ formatter = _adapters.get(tool)
82
+ if formatter is None:
83
+ return Artifact(
84
+ name="CLI Commands",
85
+ content=json.dumps({"error": f"Unknown tool: {tool}"}),
86
+ artifact_type="cli_commands",
87
+ format="json",
88
+ )
89
+
90
+ cli_name = {"github": "gh", "jira": "jira", "linear": "linear"}[tool]
91
+ available = shutil.which(cli_name) is not None
92
+ commands = formatter(artifact)
93
+ content = json.dumps({"tool": tool, "available": available, "commands": commands}, indent=2)
94
+ return Artifact(
95
+ name="CLI Commands", content=content, artifact_type="cli_commands", format="json"
96
+ )
97
+
98
+
99
+register_skill(CLIAdapterSkill())
--- a/video_processor/agent/skills/cli_adapter.py
+++ b/video_processor/agent/skills/cli_adapter.py
@@ -0,0 +1,99 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/cli_adapter.py
+++ b/video_processor/agent/skills/cli_adapter.py
@@ -0,0 +1,99 @@
1 """Skill: Push artifacts to external tools via their CLIs."""
2
3 import json
4 import shutil
5 import subprocess
6 from typing import List
7
8 from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill
9
10
11 def _format_github(artifact: Artifact) -> List[str]:
12 """Convert artifact to gh CLI commands."""
13 items = json.loads(artifact.content) if artifact.format == "json" else []
14 cmds = []
15 for item in items:
16 cmd = f"gh issue create --title {json.dumps(item.get('title', ''))}"
17 if item.get("body"):
18 cmd += f" --body {json.dumps(item['body'])}"
19 for label in item.get("labels", []):
20 cmd += f" --label {json.dumps(label)}"
21 cmds.append(cmd)
22 return cmds
23
24
25 def _format_jira(artifact: Artifact) -> List[str]:
26 """Convert artifact to jira-cli commands."""
27 items = json.loads(artifact.content) if artifact.format == "json" else []
28 return [
29 f"jira issue create --summary {json.dumps(item.get('title', ''))}"
30 f" --description {json.dumps(item.get('body', item.get('description', '')))}"
31 for item in items
32 ]
33
34
35 def _format_linear(artifact: Artifact) -> List[str]:
36 """Convert artifact to linear CLI commands."""
37 items = json.loads(artifact.content) if artifact.format == "json" else []
38 return [
39 f"linear issue create --title {json.dumps(item.get('title', ''))}"
40 f" --description {json.dumps(item.get('body', item.get('description', '')))}"
41 for item in items
42 ]
43
44
45 _adapters = {"github": _format_github, "jira": _format_jira, "linear": _format_linear}
46
47
48 def run_commands(commands: List[str], dry_run: bool = True) -> List[dict]:
49 """Execute CLI commands. In dry_run mode, just return what would run."""
50 results = []
51 for cmd in commands:
52 if dry_run:
53 results.append({"command": cmd, "status": "dry_run"})
54 else:
55 proc = subprocess.run(cmd, shell=True, capture_output=True, text=True)
56 results.append(
57 {
58 "command": cmd,
59 "returncode": proc.returncode,
60 "stdout": proc.stdout.strip(),
61 "stderr": proc.stderr.strip(),
62 }
63 )
64 return results
65
66
67 class CLIAdapterSkill(Skill):
68 name = "cli_adapter"
69 description = "Push artifacts to external tools via their CLIs"
70
71 def execute(self, context: AgentContext, **kwargs) -> Artifact:
72 tool = kwargs.get("tool", "github")
73 artifact = kwargs.get("artifact")
74 if artifact is None and context.artifacts:
75 artifact = context.artifacts[-1]
76 if artifact is None:
77 return Artifact(
78 name="CLI Commands", content="[]", artifact_type="cli_commands", format="json"
79 )
80
81 formatter = _adapters.get(tool)
82 if formatter is None:
83 return Artifact(
84 name="CLI Commands",
85 content=json.dumps({"error": f"Unknown tool: {tool}"}),
86 artifact_type="cli_commands",
87 format="json",
88 )
89
90 cli_name = {"github": "gh", "jira": "jira", "linear": "linear"}[tool]
91 available = shutil.which(cli_name) is not None
92 commands = formatter(artifact)
93 content = json.dumps({"tool": tool, "available": available, "commands": commands}, indent=2)
94 return Artifact(
95 name="CLI Commands", content=content, artifact_type="cli_commands", format="json"
96 )
97
98
99 register_skill(CLIAdapterSkill())
A video_processor/agent/skills/doc_generator.py
+76
--- a/video_processor/agent/skills/doc_generator.py
+++ b/video_processor/agent/skills/doc_generator.py
@@ -0,0 +1,76 @@
1
+"""Skill: Generate technical documentation, ADRs, or meeting notes."""
2
+
3
+from video_processor.agent.skills.base import (
4
+ AgentContext,
5
+ Artifact,
6
+ Skill,
7
+ register_skill,
8
+)
9
+
10
+_DOC_PROMPTS = {
11
+ "technical_doc": (
12
+ "Generate technical documentation with:\n"
13
+ "1. Overview\n2. Architecture\n3. Components & Interfaces\n"
14
+ "4. Data Flow\n5. Deployment & Configuration\n"
15
+ "6. API Reference (if applicable)"
16
+ ),
17
+ "adr": (
18
+ "Generate an Architecture Decision Record (ADR) with:\n"
19
+ "1. Title\n2. Status (Proposed)\n3. Context\n"
20
+ "4. Decision\n5. Consequences\n6. Alternatives Considered"
21
+ ),
22
+ "meeting_notes": (
23
+ "Generate structured meeting notes with:\n"
24
+ "1. Meeting Summary\n2. Key Discussion Points\n"
25
+ "3. Decisions Made\n4. Action Items (with owners)\n"
26
+ "5. Open Questions\n6. Next Steps"
27
+ ),
28
+}
29
+
30
+
31
+class DocGeneratorSkill(Skill):
32
+ name = "doc_generator"
33
+ description = "Generate technical documentation, ADRs, or meeting notes"
34
+
35
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
36
+ doc_type = kwargs.get("doc_type", "technical_doc")
37
+ stats = context.query_engine.stats()
38
+ entities = context.query_engine.entities()
39
+ relationships = context.query_engine.relationships()
40
+
41
+ doc_instructions = _DOC_PROMPTS.get(doc_type, _DOC_PROMPTS["technical_doc"])
42
+ doc_label = doc_type.replace("_", " ")
43
+
44
+ parts = [
45
+ f"You are a technical writer. Generate a {doc_label} "
46
+ "from the following knowledge graph context.",
47
+ "",
48
+ "## Knowledge Graph Overview",
49
+ stats.to_text(),
50
+ "",
51
+ "## Entities",
52
+ entities.to_text(),
53
+ "",
54
+ "## Relationships",
55
+ relationships.to_text(),
56
+ "",
57
+ "## Planning Entities",
58
+ ]
59
+ for e in context.planning_entities:
60
+ parts.append(f"- {e}")
61
+
62
+ parts.append(f"\n{doc_instructions}\n\nReturn ONLY the markdown.")
63
+
64
+ prompt = "\n".join(parts)
65
+ response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
66
+
67
+ return Artifact(
68
+ name=doc_label.title(),
69
+ content=response,
70
+ artifact_type="document",
71
+ format="markdown",
72
+ metadata={"doc_type": doc_type},
73
+ )
74
+
75
+
76
+register_skill(DocGeneratorSkill())
--- a/video_processor/agent/skills/doc_generator.py
+++ b/video_processor/agent/skills/doc_generator.py
@@ -0,0 +1,76 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/doc_generator.py
+++ b/video_processor/agent/skills/doc_generator.py
@@ -0,0 +1,76 @@
1 """Skill: Generate technical documentation, ADRs, or meeting notes."""
2
3 from video_processor.agent.skills.base import (
4 AgentContext,
5 Artifact,
6 Skill,
7 register_skill,
8 )
9
10 _DOC_PROMPTS = {
11 "technical_doc": (
12 "Generate technical documentation with:\n"
13 "1. Overview\n2. Architecture\n3. Components & Interfaces\n"
14 "4. Data Flow\n5. Deployment & Configuration\n"
15 "6. API Reference (if applicable)"
16 ),
17 "adr": (
18 "Generate an Architecture Decision Record (ADR) with:\n"
19 "1. Title\n2. Status (Proposed)\n3. Context\n"
20 "4. Decision\n5. Consequences\n6. Alternatives Considered"
21 ),
22 "meeting_notes": (
23 "Generate structured meeting notes with:\n"
24 "1. Meeting Summary\n2. Key Discussion Points\n"
25 "3. Decisions Made\n4. Action Items (with owners)\n"
26 "5. Open Questions\n6. Next Steps"
27 ),
28 }
29
30
31 class DocGeneratorSkill(Skill):
32 name = "doc_generator"
33 description = "Generate technical documentation, ADRs, or meeting notes"
34
35 def execute(self, context: AgentContext, **kwargs) -> Artifact:
36 doc_type = kwargs.get("doc_type", "technical_doc")
37 stats = context.query_engine.stats()
38 entities = context.query_engine.entities()
39 relationships = context.query_engine.relationships()
40
41 doc_instructions = _DOC_PROMPTS.get(doc_type, _DOC_PROMPTS["technical_doc"])
42 doc_label = doc_type.replace("_", " ")
43
44 parts = [
45 f"You are a technical writer. Generate a {doc_label} "
46 "from the following knowledge graph context.",
47 "",
48 "## Knowledge Graph Overview",
49 stats.to_text(),
50 "",
51 "## Entities",
52 entities.to_text(),
53 "",
54 "## Relationships",
55 relationships.to_text(),
56 "",
57 "## Planning Entities",
58 ]
59 for e in context.planning_entities:
60 parts.append(f"- {e}")
61
62 parts.append(f"\n{doc_instructions}\n\nReturn ONLY the markdown.")
63
64 prompt = "\n".join(parts)
65 response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
66
67 return Artifact(
68 name=doc_label.title(),
69 content=response,
70 artifact_type="document",
71 format="markdown",
72 metadata={"doc_type": doc_type},
73 )
74
75
76 register_skill(DocGeneratorSkill())
A video_processor/agent/skills/github_integration.py
+93
--- a/video_processor/agent/skills/github_integration.py
+++ b/video_processor/agent/skills/github_integration.py
@@ -0,0 +1,93 @@
1
+"""Skill: Generate GitHub issues from task breakdown artifacts."""
2
+
3
+import json
4
+import shutil
5
+import subprocess
6
+from typing import List, Optional
7
+
8
+from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill
9
+
10
+
11
+def _task_to_issue(task: dict) -> dict:
12
+ """Convert a task dict to a GitHub issue object."""
13
+ deps = task.get("dependencies", [])
14
+ body_parts = [
15
+ f"## Description\n{task.get('description', task.get('title', ''))}",
16
+ f"**Priority:** {task.get('priority', 'medium')}",
17
+ f"**Estimate:** {task.get('estimate', 'unknown')}",
18
+ ]
19
+ if deps:
20
+ body_parts.append(f"**Dependencies:** {', '.join(str(d) for d in deps)}")
21
+ labels = [task.get("priority", "medium")]
22
+ if task.get("labels"):
23
+ labels.extend(task["labels"])
24
+ return {
25
+ "title": task.get("title", "Untitled task"),
26
+ "body": "\n\n".join(body_parts),
27
+ "labels": labels,
28
+ }
29
+
30
+
31
+def push_to_github(issues_json: str, repo: str) -> Optional[List[dict]]:
32
+ """Shell out to `gh issue create` for each issue. Returns None if gh unavailable."""
33
+ if not shutil.which("gh"):
34
+ return None
35
+ issues = json.loads(issues_json)
36
+ results = []
37
+ for issue in issues:
38
+ cmd = [
39
+ "gh",
40
+ "issue",
41
+ "create",
42
+ "--repo",
43
+ repo,
44
+ "--title",
45
+ issue["title"],
46
+ "--body",
47
+ issue["body"],
48
+ ]
49
+ for label in issue.get("labels", []):
50
+ cmd.extend(["--label", label])
51
+ proc = subprocess.run(cmd, capture_output=True, text=True)
52
+ results.append(
53
+ {
54
+ "title": issue["title"],
55
+ "returncode": proc.returncode,
56
+ "stdout": proc.stdout.strip(),
57
+ "stderr": proc.stderr.strip(),
58
+ }
59
+ )
60
+ return results
61
+
62
+
63
+class GitHubIssuesSkill(Skill):
64
+ name = "github_issues"
65
+ description = "Generate GitHub issues from task breakdown"
66
+
67
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
68
+ task_artifact = next((a for a in context.artifacts if a.artifact_type == "task_list"), None)
69
+ if task_artifact:
70
+ tasks = json.loads(task_artifact.content)
71
+ else:
72
+ # Generate minimal task list inline from planning entities
73
+ tasks = [
74
+ {
75
+ "title": str(e),
76
+ "description": str(e),
77
+ "priority": "medium",
78
+ "estimate": "unknown",
79
+ }
80
+ for e in context.planning_entities
81
+ ]
82
+
83
+ issues = [_task_to_issue(t) for t in tasks]
84
+ content = json.dumps(issues, indent=2)
85
+ return Artifact(
86
+ name="GitHub Issues",
87
+ content=content,
88
+ artifact_type="issues",
89
+ format="json",
90
+ )
91
+
92
+
93
+register_skill(GitHubIssuesSkill())
--- a/video_processor/agent/skills/github_integration.py
+++ b/video_processor/agent/skills/github_integration.py
@@ -0,0 +1,93 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/github_integration.py
+++ b/video_processor/agent/skills/github_integration.py
@@ -0,0 +1,93 @@
1 """Skill: Generate GitHub issues from task breakdown artifacts."""
2
3 import json
4 import shutil
5 import subprocess
6 from typing import List, Optional
7
8 from video_processor.agent.skills.base import AgentContext, Artifact, Skill, register_skill
9
10
11 def _task_to_issue(task: dict) -> dict:
12 """Convert a task dict to a GitHub issue object."""
13 deps = task.get("dependencies", [])
14 body_parts = [
15 f"## Description\n{task.get('description', task.get('title', ''))}",
16 f"**Priority:** {task.get('priority', 'medium')}",
17 f"**Estimate:** {task.get('estimate', 'unknown')}",
18 ]
19 if deps:
20 body_parts.append(f"**Dependencies:** {', '.join(str(d) for d in deps)}")
21 labels = [task.get("priority", "medium")]
22 if task.get("labels"):
23 labels.extend(task["labels"])
24 return {
25 "title": task.get("title", "Untitled task"),
26 "body": "\n\n".join(body_parts),
27 "labels": labels,
28 }
29
30
31 def push_to_github(issues_json: str, repo: str) -> Optional[List[dict]]:
32 """Shell out to `gh issue create` for each issue. Returns None if gh unavailable."""
33 if not shutil.which("gh"):
34 return None
35 issues = json.loads(issues_json)
36 results = []
37 for issue in issues:
38 cmd = [
39 "gh",
40 "issue",
41 "create",
42 "--repo",
43 repo,
44 "--title",
45 issue["title"],
46 "--body",
47 issue["body"],
48 ]
49 for label in issue.get("labels", []):
50 cmd.extend(["--label", label])
51 proc = subprocess.run(cmd, capture_output=True, text=True)
52 results.append(
53 {
54 "title": issue["title"],
55 "returncode": proc.returncode,
56 "stdout": proc.stdout.strip(),
57 "stderr": proc.stderr.strip(),
58 }
59 )
60 return results
61
62
63 class GitHubIssuesSkill(Skill):
64 name = "github_issues"
65 description = "Generate GitHub issues from task breakdown"
66
67 def execute(self, context: AgentContext, **kwargs) -> Artifact:
68 task_artifact = next((a for a in context.artifacts if a.artifact_type == "task_list"), None)
69 if task_artifact:
70 tasks = json.loads(task_artifact.content)
71 else:
72 # Generate minimal task list inline from planning entities
73 tasks = [
74 {
75 "title": str(e),
76 "description": str(e),
77 "priority": "medium",
78 "estimate": "unknown",
79 }
80 for e in context.planning_entities
81 ]
82
83 issues = [_task_to_issue(t) for t in tasks]
84 content = json.dumps(issues, indent=2)
85 return Artifact(
86 name="GitHub Issues",
87 content=content,
88 artifact_type="issues",
89 format="json",
90 )
91
92
93 register_skill(GitHubIssuesSkill())
A video_processor/agent/skills/prd.py
+70
--- a/video_processor/agent/skills/prd.py
+++ b/video_processor/agent/skills/prd.py
@@ -0,0 +1,70 @@
1
+"""Skill: Generate a product requirements document (PRD) / feature spec."""
2
+
3
+from video_processor.agent.skills.base import (
4
+ AgentContext,
5
+ Artifact,
6
+ Skill,
7
+ register_skill,
8
+)
9
+
10
+
11
+class PRDSkill(Skill):
12
+ name = "prd"
13
+ description = "Generate a product requirements document (PRD) / feature spec"
14
+
15
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
16
+ stats = context.query_engine.stats()
17
+ entities = context.query_engine.entities()
18
+ relationships = context.query_engine.relationships()
19
+
20
+ relevant_types = {"requirement", "feature", "constraint"}
21
+ filtered = [
22
+ e for e in context.planning_entities if getattr(e, "type", "").lower() in relevant_types
23
+ ]
24
+
25
+ parts = [
26
+ "You are a product manager. Using the following knowledge "
27
+ "graph context, generate a product requirements document.",
28
+ "",
29
+ "## Knowledge Graph Overview",
30
+ stats.to_text(),
31
+ "",
32
+ "## Entities",
33
+ entities.to_text(),
34
+ "",
35
+ "## Relationships",
36
+ relationships.to_text(),
37
+ "",
38
+ "## Relevant Planning Entities",
39
+ ]
40
+ for e in filtered:
41
+ parts.append(f"- [{getattr(e, 'type', 'unknown')}] {e}")
42
+
43
+ if not filtered:
44
+ parts.append(
45
+ "(No pre-filtered entities; derive requirements from the full context above.)"
46
+ )
47
+
48
+ parts.append(
49
+ "\nGenerate a PRD with:\n"
50
+ "1. Problem Statement\n"
51
+ "2. User Stories\n"
52
+ "3. Functional Requirements\n"
53
+ "4. Non-Functional Requirements\n"
54
+ "5. Acceptance Criteria\n"
55
+ "6. Out of Scope\n\n"
56
+ "Return ONLY the markdown."
57
+ )
58
+
59
+ prompt = "\n".join(parts)
60
+ response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
61
+
62
+ return Artifact(
63
+ name="Product Requirements Document",
64
+ content=response,
65
+ artifact_type="prd",
66
+ format="markdown",
67
+ )
68
+
69
+
70
+register_skill(PRDSkill())
--- a/video_processor/agent/skills/prd.py
+++ b/video_processor/agent/skills/prd.py
@@ -0,0 +1,70 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/prd.py
+++ b/video_processor/agent/skills/prd.py
@@ -0,0 +1,70 @@
1 """Skill: Generate a product requirements document (PRD) / feature spec."""
2
3 from video_processor.agent.skills.base import (
4 AgentContext,
5 Artifact,
6 Skill,
7 register_skill,
8 )
9
10
11 class PRDSkill(Skill):
12 name = "prd"
13 description = "Generate a product requirements document (PRD) / feature spec"
14
15 def execute(self, context: AgentContext, **kwargs) -> Artifact:
16 stats = context.query_engine.stats()
17 entities = context.query_engine.entities()
18 relationships = context.query_engine.relationships()
19
20 relevant_types = {"requirement", "feature", "constraint"}
21 filtered = [
22 e for e in context.planning_entities if getattr(e, "type", "").lower() in relevant_types
23 ]
24
25 parts = [
26 "You are a product manager. Using the following knowledge "
27 "graph context, generate a product requirements document.",
28 "",
29 "## Knowledge Graph Overview",
30 stats.to_text(),
31 "",
32 "## Entities",
33 entities.to_text(),
34 "",
35 "## Relationships",
36 relationships.to_text(),
37 "",
38 "## Relevant Planning Entities",
39 ]
40 for e in filtered:
41 parts.append(f"- [{getattr(e, 'type', 'unknown')}] {e}")
42
43 if not filtered:
44 parts.append(
45 "(No pre-filtered entities; derive requirements from the full context above.)"
46 )
47
48 parts.append(
49 "\nGenerate a PRD with:\n"
50 "1. Problem Statement\n"
51 "2. User Stories\n"
52 "3. Functional Requirements\n"
53 "4. Non-Functional Requirements\n"
54 "5. Acceptance Criteria\n"
55 "6. Out of Scope\n\n"
56 "Return ONLY the markdown."
57 )
58
59 prompt = "\n".join(parts)
60 response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
61
62 return Artifact(
63 name="Product Requirements Document",
64 content=response,
65 artifact_type="prd",
66 format="markdown",
67 )
68
69
70 register_skill(PRDSkill())
A video_processor/agent/skills/project_plan.py
+74
--- a/video_processor/agent/skills/project_plan.py
+++ b/video_processor/agent/skills/project_plan.py
@@ -0,0 +1,74 @@
1
+"""Skill: Generate a structured project plan from knowledge graph."""
2
+
3
+from video_processor.agent.skills.base import (
4
+ AgentContext,
5
+ Artifact,
6
+ Skill,
7
+ register_skill,
8
+)
9
+
10
+
11
+def _group_entities_by_type(entities):
12
+ """Group planning entities by their type."""
13
+ grouped = {}
14
+ for e in entities:
15
+ etype = getattr(e, "type", "unknown")
16
+ grouped.setdefault(etype, []).append(e)
17
+ return grouped
18
+
19
+
20
+class ProjectPlanSkill(Skill):
21
+ name = "project_plan"
22
+ description = "Generate a structured project plan from knowledge graph"
23
+
24
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
25
+ stats = context.query_engine.stats()
26
+ entities = context.query_engine.entities()
27
+ relationships = context.query_engine.relationships()
28
+ grouped = _group_entities_by_type(context.planning_entities)
29
+
30
+ parts = [
31
+ "You are a project planning expert. Using the following "
32
+ "knowledge graph context, generate a comprehensive "
33
+ "project plan in markdown.",
34
+ "",
35
+ "## Knowledge Graph Overview",
36
+ stats.to_text(),
37
+ "",
38
+ "## Entities",
39
+ entities.to_text(),
40
+ "",
41
+ "## Relationships",
42
+ relationships.to_text(),
43
+ "",
44
+ "## Planning Entities (by type)",
45
+ ]
46
+ for etype, elist in grouped.items():
47
+ parts.append(f"\n### {etype}")
48
+ for e in elist:
49
+ parts.append(f"- {e}")
50
+
51
+ parts.append(
52
+ "\nGenerate a markdown project plan with:\n"
53
+ "1. Executive Summary\n"
54
+ "2. Goals & Objectives\n"
55
+ "3. Scope\n"
56
+ "4. Phases & Milestones\n"
57
+ "5. Resource Requirements\n"
58
+ "6. Risks & Mitigations\n"
59
+ "7. Success Criteria\n\n"
60
+ "Return ONLY the markdown."
61
+ )
62
+
63
+ prompt = "\n".join(parts)
64
+ response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
65
+
66
+ return Artifact(
67
+ name="Project Plan",
68
+ content=response,
69
+ artifact_type="project_plan",
70
+ format="markdown",
71
+ )
72
+
73
+
74
+register_skill(ProjectPlanSkill())
--- a/video_processor/agent/skills/project_plan.py
+++ b/video_processor/agent/skills/project_plan.py
@@ -0,0 +1,74 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/project_plan.py
+++ b/video_processor/agent/skills/project_plan.py
@@ -0,0 +1,74 @@
1 """Skill: Generate a structured project plan from knowledge graph."""
2
3 from video_processor.agent.skills.base import (
4 AgentContext,
5 Artifact,
6 Skill,
7 register_skill,
8 )
9
10
11 def _group_entities_by_type(entities):
12 """Group planning entities by their type."""
13 grouped = {}
14 for e in entities:
15 etype = getattr(e, "type", "unknown")
16 grouped.setdefault(etype, []).append(e)
17 return grouped
18
19
20 class ProjectPlanSkill(Skill):
21 name = "project_plan"
22 description = "Generate a structured project plan from knowledge graph"
23
24 def execute(self, context: AgentContext, **kwargs) -> Artifact:
25 stats = context.query_engine.stats()
26 entities = context.query_engine.entities()
27 relationships = context.query_engine.relationships()
28 grouped = _group_entities_by_type(context.planning_entities)
29
30 parts = [
31 "You are a project planning expert. Using the following "
32 "knowledge graph context, generate a comprehensive "
33 "project plan in markdown.",
34 "",
35 "## Knowledge Graph Overview",
36 stats.to_text(),
37 "",
38 "## Entities",
39 entities.to_text(),
40 "",
41 "## Relationships",
42 relationships.to_text(),
43 "",
44 "## Planning Entities (by type)",
45 ]
46 for etype, elist in grouped.items():
47 parts.append(f"\n### {etype}")
48 for e in elist:
49 parts.append(f"- {e}")
50
51 parts.append(
52 "\nGenerate a markdown project plan with:\n"
53 "1. Executive Summary\n"
54 "2. Goals & Objectives\n"
55 "3. Scope\n"
56 "4. Phases & Milestones\n"
57 "5. Resource Requirements\n"
58 "6. Risks & Mitigations\n"
59 "7. Success Criteria\n\n"
60 "Return ONLY the markdown."
61 )
62
63 prompt = "\n".join(parts)
64 response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
65
66 return Artifact(
67 name="Project Plan",
68 content=response,
69 artifact_type="project_plan",
70 format="markdown",
71 )
72
73
74 register_skill(ProjectPlanSkill())
A video_processor/agent/skills/requirements_chat.py
+95
--- a/video_processor/agent/skills/requirements_chat.py
+++ b/video_processor/agent/skills/requirements_chat.py
@@ -0,0 +1,95 @@
1
+"""Skill: Interactive requirements gathering via guided questions."""
2
+
3
+import json
4
+
5
+from video_processor.agent.skills.base import (
6
+ AgentContext,
7
+ Artifact,
8
+ Skill,
9
+ register_skill,
10
+)
11
+from video_processor.utils.json_parsing import parse_json_from_response
12
+
13
+
14
+class RequirementsChatSkill(Skill):
15
+ name = "requirements_chat"
16
+ description = "Interactive requirements gathering via guided questions"
17
+
18
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
19
+ """Generate a structured requirements questionnaire."""
20
+ stats = context.query_engine.stats()
21
+ entities = context.query_engine.entities()
22
+
23
+ parts = [
24
+ "You are a requirements analyst. Based on the following "
25
+ "knowledge graph context, generate a requirements "
26
+ "gathering questionnaire.",
27
+ "",
28
+ "## Knowledge Graph Overview",
29
+ stats.to_text(),
30
+ "",
31
+ "## Entities",
32
+ entities.to_text(),
33
+ "",
34
+ "## Planning Entities",
35
+ ]
36
+ for e in context.planning_entities:
37
+ parts.append(f"- {e}")
38
+
39
+ parts.append(
40
+ '\nGenerate a JSON object with a "questions" array. '
41
+ "Each question should have:\n"
42
+ '- "id": string (e.g. "Q1")\n'
43
+ '- "category": "goals"|"constraints"|"priorities"|"scope"\n'
44
+ '- "question": string\n'
45
+ '- "context": string (why this matters)\n\n'
46
+ "Include 8-12 targeted questions.\n\n"
47
+ "Return ONLY the JSON."
48
+ )
49
+
50
+ prompt = "\n".join(parts)
51
+ response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
52
+ parsed = parse_json_from_response(response)
53
+ content = json.dumps(parsed, indent=2) if not isinstance(parsed, str) else parsed
54
+
55
+ return Artifact(
56
+ name="Requirements Questionnaire",
57
+ content=content,
58
+ artifact_type="requirements",
59
+ format="json",
60
+ metadata={"stage": "questionnaire"},
61
+ )
62
+
63
+ def gather_requirements(self, context: AgentContext, answers: dict) -> dict:
64
+ """Take Q&A pairs and synthesize structured requirements."""
65
+ stats = context.query_engine.stats()
66
+
67
+ qa_text = ""
68
+ for qid, answer in answers.items():
69
+ qa_text += f"- {qid}: {answer}\n"
70
+
71
+ parts = [
72
+ "You are a requirements analyst. Based on the knowledge "
73
+ "graph context and the answered questions, synthesize "
74
+ "structured requirements.",
75
+ "",
76
+ "## Knowledge Graph Overview",
77
+ stats.to_text(),
78
+ "",
79
+ "## Answers",
80
+ qa_text,
81
+ "Return a JSON object with:\n"
82
+ '- "goals": list of goal strings\n'
83
+ '- "constraints": list of constraint strings\n'
84
+ '- "priorities": list (ordered high to low)\n'
85
+ '- "scope": {"in_scope": [...], "out_of_scope": [...]}\n\n'
86
+ "Return ONLY the JSON.",
87
+ ]
88
+
89
+ prompt = "\n".join(parts)
90
+ response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
91
+ result = parse_json_from_response(response)
92
+ return result if isinstance(result, dict) else {"raw": result}
93
+
94
+
95
+register_skill(RequirementsChatSkill())
--- a/video_processor/agent/skills/requirements_chat.py
+++ b/video_processor/agent/skills/requirements_chat.py
@@ -0,0 +1,95 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/requirements_chat.py
+++ b/video_processor/agent/skills/requirements_chat.py
@@ -0,0 +1,95 @@
1 """Skill: Interactive requirements gathering via guided questions."""
2
3 import json
4
5 from video_processor.agent.skills.base import (
6 AgentContext,
7 Artifact,
8 Skill,
9 register_skill,
10 )
11 from video_processor.utils.json_parsing import parse_json_from_response
12
13
14 class RequirementsChatSkill(Skill):
15 name = "requirements_chat"
16 description = "Interactive requirements gathering via guided questions"
17
18 def execute(self, context: AgentContext, **kwargs) -> Artifact:
19 """Generate a structured requirements questionnaire."""
20 stats = context.query_engine.stats()
21 entities = context.query_engine.entities()
22
23 parts = [
24 "You are a requirements analyst. Based on the following "
25 "knowledge graph context, generate a requirements "
26 "gathering questionnaire.",
27 "",
28 "## Knowledge Graph Overview",
29 stats.to_text(),
30 "",
31 "## Entities",
32 entities.to_text(),
33 "",
34 "## Planning Entities",
35 ]
36 for e in context.planning_entities:
37 parts.append(f"- {e}")
38
39 parts.append(
40 '\nGenerate a JSON object with a "questions" array. '
41 "Each question should have:\n"
42 '- "id": string (e.g. "Q1")\n'
43 '- "category": "goals"|"constraints"|"priorities"|"scope"\n'
44 '- "question": string\n'
45 '- "context": string (why this matters)\n\n'
46 "Include 8-12 targeted questions.\n\n"
47 "Return ONLY the JSON."
48 )
49
50 prompt = "\n".join(parts)
51 response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
52 parsed = parse_json_from_response(response)
53 content = json.dumps(parsed, indent=2) if not isinstance(parsed, str) else parsed
54
55 return Artifact(
56 name="Requirements Questionnaire",
57 content=content,
58 artifact_type="requirements",
59 format="json",
60 metadata={"stage": "questionnaire"},
61 )
62
63 def gather_requirements(self, context: AgentContext, answers: dict) -> dict:
64 """Take Q&A pairs and synthesize structured requirements."""
65 stats = context.query_engine.stats()
66
67 qa_text = ""
68 for qid, answer in answers.items():
69 qa_text += f"- {qid}: {answer}\n"
70
71 parts = [
72 "You are a requirements analyst. Based on the knowledge "
73 "graph context and the answered questions, synthesize "
74 "structured requirements.",
75 "",
76 "## Knowledge Graph Overview",
77 stats.to_text(),
78 "",
79 "## Answers",
80 qa_text,
81 "Return a JSON object with:\n"
82 '- "goals": list of goal strings\n'
83 '- "constraints": list of constraint strings\n'
84 '- "priorities": list (ordered high to low)\n'
85 '- "scope": {"in_scope": [...], "out_of_scope": [...]}\n\n'
86 "Return ONLY the JSON.",
87 ]
88
89 prompt = "\n".join(parts)
90 response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
91 result = parse_json_from_response(response)
92 return result if isinstance(result, dict) else {"raw": result}
93
94
95 register_skill(RequirementsChatSkill())
A video_processor/agent/skills/roadmap.py
+68
--- a/video_processor/agent/skills/roadmap.py
+++ b/video_processor/agent/skills/roadmap.py
@@ -0,0 +1,68 @@
1
+"""Skill: Generate a product/project roadmap."""
2
+
3
+from video_processor.agent.skills.base import (
4
+ AgentContext,
5
+ Artifact,
6
+ Skill,
7
+ register_skill,
8
+)
9
+
10
+
11
+class RoadmapSkill(Skill):
12
+ name = "roadmap"
13
+ description = "Generate a product/project roadmap"
14
+
15
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
16
+ stats = context.query_engine.stats()
17
+ entities = context.query_engine.entities()
18
+ relationships = context.query_engine.relationships()
19
+
20
+ roadmap_types = {"milestone", "feature", "dependency"}
21
+ relevant = [
22
+ e for e in context.planning_entities if getattr(e, "type", "").lower() in roadmap_types
23
+ ]
24
+
25
+ parts = [
26
+ "You are a product strategist. Using the following "
27
+ "knowledge graph context, generate a product roadmap.",
28
+ "",
29
+ "## Knowledge Graph Overview",
30
+ stats.to_text(),
31
+ "",
32
+ "## Entities",
33
+ entities.to_text(),
34
+ "",
35
+ "## Relationships",
36
+ relationships.to_text(),
37
+ "",
38
+ "## Milestones, Features & Dependencies",
39
+ ]
40
+ for e in relevant:
41
+ parts.append(f"- [{getattr(e, 'type', 'unknown')}] {e}")
42
+
43
+ if not relevant:
44
+ parts.append(
45
+ "(No pre-filtered entities; derive roadmap items from the full context above.)"
46
+ )
47
+
48
+ parts.append(
49
+ "\nGenerate a markdown roadmap with:\n"
50
+ "1. Vision & Strategy\n"
51
+ "2. Phases (with timeline estimates)\n"
52
+ "3. Key Dependencies\n"
53
+ "4. A Mermaid Gantt chart summarizing the timeline\n\n"
54
+ "Return ONLY the markdown."
55
+ )
56
+
57
+ prompt = "\n".join(parts)
58
+ response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
59
+
60
+ return Artifact(
61
+ name="Roadmap",
62
+ content=response,
63
+ artifact_type="roadmap",
64
+ format="markdown",
65
+ )
66
+
67
+
68
+register_skill(RoadmapSkill())
--- a/video_processor/agent/skills/roadmap.py
+++ b/video_processor/agent/skills/roadmap.py
@@ -0,0 +1,68 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/roadmap.py
+++ b/video_processor/agent/skills/roadmap.py
@@ -0,0 +1,68 @@
1 """Skill: Generate a product/project roadmap."""
2
3 from video_processor.agent.skills.base import (
4 AgentContext,
5 Artifact,
6 Skill,
7 register_skill,
8 )
9
10
11 class RoadmapSkill(Skill):
12 name = "roadmap"
13 description = "Generate a product/project roadmap"
14
15 def execute(self, context: AgentContext, **kwargs) -> Artifact:
16 stats = context.query_engine.stats()
17 entities = context.query_engine.entities()
18 relationships = context.query_engine.relationships()
19
20 roadmap_types = {"milestone", "feature", "dependency"}
21 relevant = [
22 e for e in context.planning_entities if getattr(e, "type", "").lower() in roadmap_types
23 ]
24
25 parts = [
26 "You are a product strategist. Using the following "
27 "knowledge graph context, generate a product roadmap.",
28 "",
29 "## Knowledge Graph Overview",
30 stats.to_text(),
31 "",
32 "## Entities",
33 entities.to_text(),
34 "",
35 "## Relationships",
36 relationships.to_text(),
37 "",
38 "## Milestones, Features & Dependencies",
39 ]
40 for e in relevant:
41 parts.append(f"- [{getattr(e, 'type', 'unknown')}] {e}")
42
43 if not relevant:
44 parts.append(
45 "(No pre-filtered entities; derive roadmap items from the full context above.)"
46 )
47
48 parts.append(
49 "\nGenerate a markdown roadmap with:\n"
50 "1. Vision & Strategy\n"
51 "2. Phases (with timeline estimates)\n"
52 "3. Key Dependencies\n"
53 "4. A Mermaid Gantt chart summarizing the timeline\n\n"
54 "Return ONLY the markdown."
55 )
56
57 prompt = "\n".join(parts)
58 response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
59
60 return Artifact(
61 name="Roadmap",
62 content=response,
63 artifact_type="roadmap",
64 format="markdown",
65 )
66
67
68 register_skill(RoadmapSkill())
A video_processor/agent/skills/task_breakdown.py
+76
--- a/video_processor/agent/skills/task_breakdown.py
+++ b/video_processor/agent/skills/task_breakdown.py
@@ -0,0 +1,76 @@
1
+"""Skill: Break down goals into tasks with dependencies."""
2
+
3
+from video_processor.agent.skills.base import (
4
+ AgentContext,
5
+ Artifact,
6
+ Skill,
7
+ register_skill,
8
+)
9
+from video_processor.utils.json_parsing import parse_json_from_response
10
+
11
+
12
+class TaskBreakdownSkill(Skill):
13
+ name = "task_breakdown"
14
+ description = "Break down goals into tasks with dependencies"
15
+
16
+ def execute(self, context: AgentContext, **kwargs) -> Artifact:
17
+ stats = context.query_engine.stats()
18
+ entities = context.query_engine.entities()
19
+ relationships = context.query_engine.relationships()
20
+
21
+ task_types = {"goal", "feature", "milestone"}
22
+ relevant = [
23
+ e for e in context.planning_entities if getattr(e, "type", "").lower() in task_types
24
+ ]
25
+
26
+ parts = [
27
+ "You are a project manager. Using the following knowledge "
28
+ "graph context, decompose goals and features into tasks.",
29
+ "",
30
+ "## Knowledge Graph Overview",
31
+ stats.to_text(),
32
+ "",
33
+ "## Entities",
34
+ entities.to_text(),
35
+ "",
36
+ "## Relationships",
37
+ relationships.to_text(),
38
+ "",
39
+ "## Goals, Features & Milestones",
40
+ ]
41
+ for e in relevant:
42
+ parts.append(f"- [{getattr(e, 'type', 'unknown')}] {e}")
43
+
44
+ if not relevant:
45
+ parts.append("(No pre-filtered entities; derive tasks from the full context above.)")
46
+
47
+ parts.append(
48
+ "\nReturn a JSON array of task objects with:\n"
49
+ '- "id": string (e.g. "T1", "T2")\n'
50
+ '- "title": string\n'
51
+ '- "description": string\n'
52
+ '- "depends_on": list of task id strings\n'
53
+ '- "priority": "high" | "medium" | "low"\n'
54
+ '- "estimate": string (e.g. "2d", "1w")\n'
55
+ '- "assignee_role": string\n\n'
56
+ "Return ONLY the JSON."
57
+ )
58
+
59
+ prompt = "\n".join(parts)
60
+ response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
61
+ parsed = parse_json_from_response(response)
62
+
63
+ import json
64
+
65
+ content = json.dumps(parsed, indent=2) if isinstance(parsed, list) else response
66
+
67
+ return Artifact(
68
+ name="Task Breakdown",
69
+ content=content,
70
+ artifact_type="task_list",
71
+ format="json",
72
+ metadata={"tasks": parsed if isinstance(parsed, list) else []},
73
+ )
74
+
75
+
76
+register_skill(TaskBreakdownSkill())
--- a/video_processor/agent/skills/task_breakdown.py
+++ b/video_processor/agent/skills/task_breakdown.py
@@ -0,0 +1,76 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/agent/skills/task_breakdown.py
+++ b/video_processor/agent/skills/task_breakdown.py
@@ -0,0 +1,76 @@
1 """Skill: Break down goals into tasks with dependencies."""
2
3 from video_processor.agent.skills.base import (
4 AgentContext,
5 Artifact,
6 Skill,
7 register_skill,
8 )
9 from video_processor.utils.json_parsing import parse_json_from_response
10
11
12 class TaskBreakdownSkill(Skill):
13 name = "task_breakdown"
14 description = "Break down goals into tasks with dependencies"
15
16 def execute(self, context: AgentContext, **kwargs) -> Artifact:
17 stats = context.query_engine.stats()
18 entities = context.query_engine.entities()
19 relationships = context.query_engine.relationships()
20
21 task_types = {"goal", "feature", "milestone"}
22 relevant = [
23 e for e in context.planning_entities if getattr(e, "type", "").lower() in task_types
24 ]
25
26 parts = [
27 "You are a project manager. Using the following knowledge "
28 "graph context, decompose goals and features into tasks.",
29 "",
30 "## Knowledge Graph Overview",
31 stats.to_text(),
32 "",
33 "## Entities",
34 entities.to_text(),
35 "",
36 "## Relationships",
37 relationships.to_text(),
38 "",
39 "## Goals, Features & Milestones",
40 ]
41 for e in relevant:
42 parts.append(f"- [{getattr(e, 'type', 'unknown')}] {e}")
43
44 if not relevant:
45 parts.append("(No pre-filtered entities; derive tasks from the full context above.)")
46
47 parts.append(
48 "\nReturn a JSON array of task objects with:\n"
49 '- "id": string (e.g. "T1", "T2")\n'
50 '- "title": string\n'
51 '- "description": string\n'
52 '- "depends_on": list of task id strings\n'
53 '- "priority": "high" | "medium" | "low"\n'
54 '- "estimate": string (e.g. "2d", "1w")\n'
55 '- "assignee_role": string\n\n'
56 "Return ONLY the JSON."
57 )
58
59 prompt = "\n".join(parts)
60 response = context.provider_manager.chat(messages=[{"role": "user", "content": prompt}])
61 parsed = parse_json_from_response(response)
62
63 import json
64
65 content = json.dumps(parsed, indent=2) if isinstance(parsed, list) else response
66
67 return Artifact(
68 name="Task Breakdown",
69 content=content,
70 artifact_type="task_list",
71 format="json",
72 metadata={"tasks": parsed if isinstance(parsed, list) else []},
73 )
74
75
76 register_skill(TaskBreakdownSkill())
M video_processor/cli/commands.py
+7 -7
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -653,10 +653,12 @@
653653
654654
 planopticon agent -I --kb ./videos --kb ./docs
655655
656656
 planopticon agent "Generate a PRD" --export ./output
657657
 """
658
+ # Ensure all skills are registered
659
+ import video_processor.agent.skills # noqa: F401
658660
 from video_processor.agent.agent_loop import PlanningAgent
659661
 from video_processor.agent.kb_context import KBContext
660662
 from video_processor.agent.skills.base import AgentContext
661663
662664
 # Build provider manager
@@ -737,18 +739,16 @@
737739
 for artifact in artifacts:
738740
 click.echo(f"\n--- {artifact.name} ({artifact.artifact_type}) ---\n")
739741
 click.echo(artifact.content)
740742
741743
 if export:
744
+ from video_processor.agent.skills.artifact_export import export_artifacts
745
+
742746
 export_dir = Path(export)
743
- export_dir.mkdir(parents=True, exist_ok=True)
744
- for artifact in artifacts:
745
- ext = ".md" if artifact.format == "markdown" else f".{artifact.format}"
746
- safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in artifact.name)
747
- fpath = export_dir / f"{safe_name}{ext}"
748
- fpath.write_text(artifact.content)
749
- click.echo(f"Exported: {fpath}")
747
+ export_artifacts(artifacts, export_dir)
748
+ click.echo(f"Exported {len(artifacts)} artifacts to {export_dir}/")
749
+ click.echo(f"Manifest: {export_dir / 'manifest.json'}")
750750
 else:
751751
 click.echo("Provide a request or use -I for interactive mode.")
752752
 click.echo("Example: planopticon agent 'Create a project plan' --kb ./results")
753753
754754
755755
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -653,10 +653,12 @@
653
654 planopticon agent -I --kb ./videos --kb ./docs
655
656 planopticon agent "Generate a PRD" --export ./output
657 """
 
 
658 from video_processor.agent.agent_loop import PlanningAgent
659 from video_processor.agent.kb_context import KBContext
660 from video_processor.agent.skills.base import AgentContext
661
662 # Build provider manager
@@ -737,18 +739,16 @@
737 for artifact in artifacts:
738 click.echo(f"\n--- {artifact.name} ({artifact.artifact_type}) ---\n")
739 click.echo(artifact.content)
740
741 if export:
 
 
742 export_dir = Path(export)
743 export_dir.mkdir(parents=True, exist_ok=True)
744 for artifact in artifacts:
745 ext = ".md" if artifact.format == "markdown" else f".{artifact.format}"
746 safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in artifact.name)
747 fpath = export_dir / f"{safe_name}{ext}"
748 fpath.write_text(artifact.content)
749 click.echo(f"Exported: {fpath}")
750 else:
751 click.echo("Provide a request or use -I for interactive mode.")
752 click.echo("Example: planopticon agent 'Create a project plan' --kb ./results")
753
754
755
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -653,10 +653,12 @@
653
654 planopticon agent -I --kb ./videos --kb ./docs
655
656 planopticon agent "Generate a PRD" --export ./output
657 """
658 # Ensure all skills are registered
659 import video_processor.agent.skills # noqa: F401
660 from video_processor.agent.agent_loop import PlanningAgent
661 from video_processor.agent.kb_context import KBContext
662 from video_processor.agent.skills.base import AgentContext
663
664 # Build provider manager
@@ -737,18 +739,16 @@
739 for artifact in artifacts:
740 click.echo(f"\n--- {artifact.name} ({artifact.artifact_type}) ---\n")
741 click.echo(artifact.content)
742
743 if export:
744 from video_processor.agent.skills.artifact_export import export_artifacts
745
746 export_dir = Path(export)
747 export_artifacts(artifacts, export_dir)
748 click.echo(f"Exported {len(artifacts)} artifacts to {export_dir}/")
749 click.echo(f"Manifest: {export_dir / 'manifest.json'}")
 
 
 
 
750 else:
751 click.echo("Provide a request or use -I for interactive mode.")
752 click.echo("Example: planopticon agent 'Create a project plan' --kb ./results")
753
754
755

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button