From 6aa642bc42b0febefd0a9f53e7614302e90328d4 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Fri, 27 Feb 2026 22:16:43 -0500 Subject: [PATCH] feat(mma): Implement tiered context scoping and add get_definition tool --- .gemini/tools.json | Bin 11872 -> 6406 bytes aggregate.py | 81 +++++++++++++++++++++++++++++++++++ mcp_client.py | 80 +++++++++++++++++++++++++++++++++- tests/test_tiered_context.py | 36 ++++++++++++++++ 4 files changed, 196 insertions(+), 1 deletion(-) create mode 100644 tests/test_tiered_context.py diff --git a/.gemini/tools.json b/.gemini/tools.json index 0a1b7092407af9d7f3d7e2139c486a1bed23958f..8b44f602917ed7e0972cf2198a2b2eb58ff2cfd0 100644 GIT binary patch literal 6406 zcmds5O>f&q5WVMD3=9-KST5C|)BC~ie@ncbxowMc(^-^`LU z?O3T{$3W79ZJONAnKy6V%-csqf8&Q}uQqGmJD@$!CZ89!;=O75kaNhqZK8Fxd_Tc= ziX~IgRu!F}9sck$Wv-4~M{)%jlDV|8v9z_0Ggaj_oKxf7pPWT%Of18U&I8kuqi!`1 z?Rste7IS@w6dmQ9*&NH1jIBIvT}w4%kzwO*%UOTQ#&}#HvJbxnSD$7juMLiyE{E&I zZKKz73pqqL3-58NhJCagMx*6DPI=|cXpIoPt(U_ZYQAe$dbS9zY9o{La1)2tZaz$94lne z`E75mYsWppAaz_IQ|A=gMLz?E>78k6>V4UN-OlK=belkeo+*iuXF6eAu_(_;27 zPVNRIg4~%uG?nwXK2X%70yb^4*n0lzJu1lPy*SuGUWlb1)Rlpa&q|G$u)~q3I)nr=VJgmv(|?WyD~4}mMf;olP6D~%$n`ws$&Za z@T7egTeumd5I`Fe1OkwL^KaM1$pfA7#=h>uH8F=+>K^3ra^$Ef@oYJR!NuCs&hXK# zEBPn*?5TRfhspRf9WmUBL~c0wt$5`+ckck?OOAkE=GKh?_m|vyM2XEnF^Tc7rVQ6c zWkdn21^zV0^i)rWz|rwGmaaY^yHJFI|p@qOCK@R(QnJ5KSGCZ)Eb> zA8(EhD4GShcg5T8?Hl7)x!YjT5@d|6@%|4kcdrk-7w8jrw~YR$Os}0=Zd~B$V}oNW7H9NP>%U(u~d| z|9D200goY-_1Mug)uV=jUDif2QA&K0rL3piWhw0gRSQLFQ<)57hItE@_sNnbv)Sw# zH@0;SL;;LBtzF&%ffl047rKbZr7`wZA&9;L%qw0)TR}m%`fDZ5<55ZkPbG6dG%_-m z^BhK#5dADHO3Pjnr_D@7-Gqb=S>h&a44l$C7fx5lC5lo0?{J|6;Z3~+6lBQ4TP{Xu z8Ic9J0a7koRa<1`Mo!YHAH{?Le0gatzbC0qz$Lcj^BeG3>E>`MF44<7KvokK4w zfRM)z&Ko&Hj94vNc2mX>X2G>S6QTzmG$RdWS14TOl&NT$lxaZp{T3ayfL+#-4!D5B ztQ17CglHCtFRql`Hl;)4Yl+k)h*j-2cx|GjiwR~|&aGrPGRwE%>zm?zq$#?xvaiFm zR_bf9Thvi!u$f9nctz}|?hqgJktrD$l#Zyo=&Fp#b5R7pLvPNKqvo=}L^W}z5$=k< z>-V~~P+D54!HQxBPn~o*Np6UtBt!MQT{VsS4wc4LLP?5*GT=-YXh0s~?RriD9#I7< z_Pz-c(IB~U^HD|_d17G#&tZpq2<~671W|jP2+e`U#m#vN5g_)O3PViZ-HJoznTqQI~5&ZJUNgxj!$Bwvs@v zI<{bxwDZZD0nc|eBUxUD3?v8yO3FTd0pxbG?b2TE8ZE*w8L|UVn!uZ zt+FD@b*Yyw8+q^r?F&c454ohpS7&dIP(3XJkT2~!?dO;RDwZyBGB^coFTOzS1)Ymj zhCYOJ0t)ZeDG~DG!Tbbh+|x^`w~6~BuGNlraFv|mR*sS~r?qpmA=OTG_IWuQYSV<$ zX}48VVp3Ri{+dQhh)D4^LonGot6CKxhsUQ=I+Q*somu?aUkCK)o*$nhY!NuL|6SLz zdtEmWr$=U~Lq^VqIdW^3wS)JuUqENK<`9G$FWB2+3*ROsqjBnw=OjS zAwOTWQwG@=(G%s|#wo~7M2O~)Tq@Zv4{OQLr*>42A|xRaomU0ef7lYJFQAbE^Y@#P zY0(8jP@$-)(1r~$*MdU2w-J>la+NkFq}UakJMBgoWGZZssP_vSeP{&iB~#rq1V1%B W?!X_=kc@ZfI&^3HG$0?nd-N~w=4HzO literal 11872 zcmeI2-ER|D5XJ9vrT!0Aq$<$TpgvTp`bKH_sDg+Hd8(?&&KE9;liCiDR`gGAdw%zL zeDChY4naTwMUmt6?%la_=gyotGrRx&`&;@leUW};v`M=|5$QD!jqYVo~}I8aZi#?CFxK{Tj^=Km%dFu z=y)gJHPwibM)ve;U;q31#LT_)QnHaU)R-c#pQRn^>Yg<1Nr$oS{zJcZvZND9;jT$q z)733~qs4*#Ppfrp>-?$Kex%Xkbfyu^JJwmWnMz_$(vdmL-)O^nysELTo@vafzDGK1 zO}Z;?u5wLQYFiU+$NI1fSOX1y)d!nyYIf#CN87i#f8N8XG&#|z*n#6Tm|LSgpYO26 zxM!?6hx(o@lplB4ri-*-lt;F;!Jc@^NIP(zHs#LcesxEy!YYTdId;Nh*abZHQ1)RR zFY@`=|JcQ{EB5F(AHL46M0|uce829TecjqUuNXOhN-MkvKd_BT%+)eiv{PZp_ME>r z<}0(;vo`m?8^4Fr0*h};59>orV5zf?^;fcAFOKB(nXD3#6IrFNoh0groE5d>Bbx(ZqMewseP$68>YXs6Ls|C16U^ZelNY! zPM{5Y##v)k8%ylHrx{sJ^VsVcZX#IIyDuBV5e=~~u@7C}h!Nr8pHtM5p~wQDeaXCF za?{oqMeJrJhp06|Mn1*+$t2S}j}h4!_aH~YK-aw=&O^j(GXq~Pj!z}+BKM z2sOxXO{|Xj(2)Ioo?~(~eXYOS`m9OLSz28h(X7(QRIhiS`+Ac0N~?xxFf*AovPU3H zd*#bm(* zT}>>jXQ&&H4I)x)1r}57fPtlAYS#}v)lWwC4IvxZ5;UWB8)gl;h73GsN1oExF|p(` zt}(c0cXo!EMj9DaT-=FNxDSYrJh)WM{4_mIkJ5Ucl@SBY1BU{&&st+2>H1h#b~WeA znWk8;tlF|4YZlJqzpREVKwV|qqa*hEMjsg3o|ryN^JUg6Dm)~Z;|+Ap*$)X}<&s&0 zOEl}dq3*e^>T0Bq;}u!ISz~^@Rb5@x;|(C|)u7?~L^=3SL_eJjVvi^!LyRuX8*8j< z%j(Tzt?tp(=u_%LP~z!43W>XE#!0Xw>c2KFmAb8sMPTYsSrLwAIpA1f15ve6=hCY= z7LL_^X)GI5SWCs%*C@;B&Ns-&*sfv8!E5<%IRgcyKol^{$Vr`ilx5NZ1B*a4Gm-sD zlVI74@`ACk&I?9+WP<6)bO(gp$KRWo*c|A(%S6X>d4;->?f6VPMNj*o6Eulwk?Pa*qy;D z;Yq9>{=l8^OQV>p%z@6cHaNq1*aH@JnTfs1-NGHA_MvdxRr$%>Cgz+`#lrGILb|AR zAF3zbbsozJE;$iXa`zfZJ6VQ1UEQ6f{=deFx5tUD@6xYy*MocrQhU$gDb4^ZVPfwHd`<#La8Z@wVhbu_j;BK(=} zx6Rstq0#L-Z~qtN@g{=`ZB~T3)YN22`^VMwlYW|M;tCf}E#1u<(y(-t9;9E=o%Hj( zZxp}(N#D)RIX+6S<0!{#FZ=vN|3RgmB;fUZA=jaznY`qDwwglXvfk4oi$*`->Mi;} z>!UMqzD19P*U=||!+c&y&C~ROyt?kZwqx?5INZ$k^K*=-&fr%_?#XNLPnZ;*1ebX} z|GqCih=(o0dHf89^u0c{hTbW1X?SGdFY}XgsFBS7wj_Ohb{t%AB7Jy^0j?B%uDjXm z(bQ{jPpQb=&tv_phKvN8bsT^Q_Ma~Mf?1hEe!9F1-U^%NSu$O^=F{lFNF(<9>KM^q zpptHSLD3BBemM42=Izeu`Q ziJWckr47AT!_|8l>vzrQsL(|OOK<3#m}SIWjos2W+1eS4>tD^TM&5j=Sxz*DRjuc7 zMar_83I>kUk=71^$NDDH*iVSa)-X#mQr_rdb^Ezbf{o~N##0*lzU{h<5!^Krvt=Z8 z3%bd?ll>f{BT)6 zf0AX{&o}aZVqf0qvOK)Myf<1pZ-r~&5$$)dO2y^+ZJB=|9S%m0!JFA}SRlt-iT6GD z9X)|~a>CqEG1i^FcGZZF_^D-Zk(XK4bIs& str: return "## Discussion History\n\n" + build_discussion_section(history) +def build_tier1_context(file_items: list[dict], screenshot_base_dir: Path, screenshots: list[str], history: list[str]) -> str: + """ + Tier 1 Context: Strategic/Orchestration. + Full content for core conductor files, summaries for others. + """ + core_files = {"product.md", "tech-stack.md", "workflow.md", "tracks.md"} + + parts = [] + + # Files section + if file_items: + sections = [] + for item in file_items: + path = item.get("path") + name = path.name if path else "" + + if name in core_files: + # Include in full + sections.append("### `" + (item.get("entry") or str(path)) + "`\n\n" + + f"```{path.suffix.lstrip('.') if path.suffix else 'text'}\n{item.get('content', '')}\n```") + else: + # Summarize + sections.append(summarize.summarise_file(path, item.get("content", ""))) + + parts.append("## Files (Tier 1 - Mixed)\n\n" + "\n\n---\n\n".join(sections)) + + if screenshots: + parts.append("## Screenshots\n\n" + build_screenshots_section(screenshot_base_dir, screenshots)) + + if history: + parts.append("## Discussion History\n\n" + build_discussion_section(history)) + + return "\n\n---\n\n".join(parts) + + +def build_tier2_context(file_items: list[dict], screenshot_base_dir: Path, screenshots: list[str], history: list[str]) -> str: + """ + Tier 2 Context: Architectural/Tech Lead. + Full content for all files (standard behavior). + """ + return build_markdown_from_items(file_items, screenshot_base_dir, screenshots, history, summary_only=False) + + +def build_tier3_context(file_items: list[dict], screenshot_base_dir: Path, screenshots: list[str], history: list[str], focus_files: list[str]) -> str: + """ + Tier 3 Context: Execution/Worker. + Full content for focus_files, summaries for others. + """ + parts = [] + + if file_items: + sections = [] + for item in file_items: + path = item.get("path") + entry = item.get("entry", "") + path_str = str(path) if path else "" + + # Check if this file is in focus_files (by name or path) + is_focus = False + for focus in focus_files: + if focus == entry or (path and focus == path.name) or focus in path_str: + is_focus = True + break + + if is_focus: + sections.append("### `" + (entry or path_str) + "`\n\n" + + f"```{path.suffix.lstrip('.') if path and path.suffix else 'text'}\n{item.get('content', '')}\n```") + else: + sections.append(summarize.summarise_file(path, item.get("content", ""))) + + parts.append("## Files (Tier 3 - Focused)\n\n" + "\n\n---\n\n".join(sections)) + + if screenshots: + parts.append("## Screenshots\n\n" + build_screenshots_section(screenshot_base_dir, screenshots)) + + if history: + parts.append("## Discussion History\n\n" + build_discussion_section(history)) + + return "\n\n---\n\n".join(parts) + + def build_markdown(base_dir: Path, files: list[str], screenshot_base_dir: Path, screenshots: list[str], history: list[str], summary_only: bool = False) -> str: parts = [] # STATIC PREFIX: Files and Screenshots must go first to maximize Cache Hits diff --git a/mcp_client.py b/mcp_client.py index c42356c..7d31019 100644 --- a/mcp_client.py +++ b/mcp_client.py @@ -281,6 +281,60 @@ def get_code_outline(path: str) -> str: return f"ERROR generating outline for '{path}': {e}" +def get_definition(path: str, name: str) -> str: + """ + Returns the source code for a specific class, function, or method definition. + path: Path to the code file. + name: Name of the definition to retrieve (e.g., 'MyClass', 'my_function', 'MyClass.my_method'). + """ + p, err = _resolve_and_check(path) + if err: + return err + if not p.exists(): + return f"ERROR: file not found: {path}" + if not p.is_file(): + return f"ERROR: not a file: {path}" + + if p.suffix != ".py": + return f"ERROR: get_definition currently only supports .py files (unsupported: {p.suffix})" + + try: + import ast + code = p.read_text(encoding="utf-8") + lines = code.splitlines() + tree = ast.parse(code) + + # Split name for methods (e.g., "MyClass.my_method") + parts = name.split(".") + target_class = parts[0] if len(parts) > 1 else None + target_name = parts[-1] + + def get_source_from_node(node): + # In Python 3.8+, ast.get_source_segment is available + # But we can also use lineno and end_lineno + if hasattr(node, "lineno") and hasattr(node, "end_lineno"): + # lineno is 1-indexed + start = node.lineno - 1 + end = node.end_lineno + return "\n".join(lines[start:end]) + return f"ERROR: Could not extract source for node {node}" + + for node in ast.walk(tree): + if target_class: + if isinstance(node, ast.ClassDef) and node.name == target_class: + for body_node in node.body: + if isinstance(body_node, (ast.FunctionDef, ast.AsyncFunctionDef)) and body_node.name == target_name: + return get_source_from_node(body_node) + else: + if isinstance(node, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == target_name: + return get_source_from_node(node) + + return f"ERROR: could not find definition '{name}' in {path}" + + except Exception as e: + return f"ERROR retrieving definition '{name}' from '{path}': {e}" + + def get_git_diff(path: str, base_rev: str = "HEAD", head_rev: str = "") -> str: """ Returns the git diff for a file or directory. @@ -436,7 +490,7 @@ def get_ui_performance() -> str: # ------------------------------------------------------------------ tool dispatch -TOOL_NAMES = {"read_file", "list_directory", "search_files", "get_file_summary", "get_python_skeleton", "get_code_outline", "get_git_diff", "web_search", "fetch_url", "get_ui_performance"} +TOOL_NAMES = {"read_file", "list_directory", "search_files", "get_file_summary", "get_python_skeleton", "get_code_outline", "get_definition", "get_git_diff", "web_search", "fetch_url", "get_ui_performance"} def dispatch(tool_name: str, tool_input: dict) -> str: @@ -455,6 +509,8 @@ def dispatch(tool_name: str, tool_input: dict) -> str: return get_python_skeleton(tool_input.get("path", "")) if tool_name == "get_code_outline": return get_code_outline(tool_input.get("path", "")) + if tool_name == "get_definition": + return get_definition(tool_input.get("path", ""), tool_input.get("name", "")) if tool_name == "get_git_diff": return get_git_diff( tool_input.get("path", ""), @@ -586,6 +642,27 @@ MCP_TOOL_SPECS = [ "required": ["path"], }, }, + { + "name": "get_definition", + "description": ( + "Get the full source code of a specific class, function, or method definition. " + "This is more efficient than reading the whole file if you know what you're looking for." + ), + "parameters": { + "type": "object", + "properties": { + "path": { + "type": "string", + "description": "Path to the .py file.", + }, + "name": { + "type": "string", + "description": "The name of the class or function to retrieve. Use 'ClassName.method_name' for methods.", + } + }, + "required": ["path", "name"], + }, + }, { "name": "get_git_diff", "description": ( @@ -648,3 +725,4 @@ MCP_TOOL_SPECS = [ } } ] + diff --git a/tests/test_tiered_context.py b/tests/test_tiered_context.py new file mode 100644 index 0000000..6ad8712 --- /dev/null +++ b/tests/test_tiered_context.py @@ -0,0 +1,36 @@ +import pytest +from pathlib import Path +from aggregate import build_tier1_context, build_tier2_context, build_tier3_context + +def test_build_tier1_context_exists(): + # This should fail if the function is not defined + file_items = [ + {"path": Path("conductor/product.md"), "entry": "conductor/product.md", "content": "Product content", "error": False}, + {"path": Path("other.py"), "entry": "other.py", "content": "Other content", "error": False} + ] + history = ["User: hello", "AI: hi"] + + result = build_tier1_context(file_items, Path("."), [], history) + + assert "Product content" in result + # other.py should be summarized, not full content in a code block + assert "Other content" not in result or "Summarized" in result # Assuming summary format + +def test_build_tier2_context_exists(): + file_items = [ + {"path": Path("other.py"), "entry": "other.py", "content": "Other content", "error": False} + ] + history = ["User: hello"] + result = build_tier2_context(file_items, Path("."), [], history) + assert "Other content" in result + +def test_build_tier3_context_exists(): + file_items = [ + {"path": Path("focus.py"), "entry": "focus.py", "content": "Focus content", "error": False}, + {"path": Path("other.py"), "entry": "other.py", "content": "Other content", "error": False} + ] + history = ["User: hello"] + result = build_tier3_context(file_items, Path("."), [], history, focus_files=["focus.py"]) + + assert "Focus content" in result + assert "Other content" not in result