dialoghelper.utils

from dialoghelper import *

from fastcore import tools
from fastcore.test import *

ast-grep

---

source

ast_py

def ast_py(
    code:str
):

Get an SgRoot root node for python code

node = ast_py("print('hello world')")
stmt = node.find(pattern="print($A)")
res = stmt.get_match('A')
res.text(),res.range()

("'hello world'",
 Range(start=Pos(line=0, col=6, index=6), end=Pos(line=0, col=19, index=19)))

---

source

ast_grep

def ast_grep(
    pattern:str, # ast-grep pattern to search, e.g "post($A, data=$B, $$$)"
    path:str='.', # path to recursively search for files
    lang:str='python', # language to search/scan
):

Use ast-grep to find code patterns by AST structure (not text).

Pattern syntax: - $VAR captures single nodes, \[$ captures multiple - Match structure directly: `def $FUNC(\]$)finds any function;class $CLASSfinds classes regardless of inheritance - DON'T include:` - it’s concrete syntax, not AST structure - Whitespace/formatting ignored - matches structural equivalence

Examples: import $MODULE (find imports); $OBJ.$METHOD($$$) (find method calls); await $EXPR (find await expressions)

Useful for: Refactoring—find all uses of deprecated APIs or changed signatures; Security review—locate SQL queries, file operations, eval calls; Code exploration—understand how libraries are used across codebase; Pattern analysis—find async functions, error handlers, decorators; Better than regex—handles multi-line code, nested structures, respects syntax

The ast_grep function calls the ast-grep CLI, which is used for searching code based on its structure rather than just text patterns. Unlike regular expressions that match character sequences, ast-grep understands the syntax of programming languages and lets you search for code patterns in a way that respects the language’s grammar. This means you can find function calls, variable assignments, or other code constructs even when they’re formatted differently or have varying amounts of whitespace.

The key advantage is using metavariables (like $A, $B, $$$) as placeholders in your search patterns. When you search for xpost($A, data=$B, $$$), you’re asking to find all calls to xpost where the first argument can be anything (captured as $A), there’s a keyword argument data with any value (captured as $B), and there may be additional arguments after that (the $$$ matches zero or more remaining arguments). This is much more reliable than trying to write a regex that handles all the variations of how that function might be called.

In the example below, we search for calls to xpost in the parent directory and extract both the matched code and the specific values of our metavariables, showing us exactly where and how this function is being used in the codebase.

res = ast_grep(r"xpost($A, data=$B, $$$)", '..')
[(o['text'],o['metaVariables']['single'],o['file']) for o in res]

[('xpost(url, data=data, headers=headers, timeout=timeout)',
  {'A': {'text': 'url',
    'range': {'byteOffset': {'start': 5948, 'end': 5951},
     'start': {'line': 128, 'column': 30},
     'end': {'line': 128, 'column': 33}}},
   'B': {'text': 'data',
    'range': {'byteOffset': {'start': 5958, 'end': 5962},
     'start': {'line': 128, 'column': 40},
     'end': {'line': 128, 'column': 44}}}},
  'dialoghelper/core.py')]

Basic Patterns: - Match code structure directly: console.log($ARG) - Metavariables capture parts: $VAR (single), $$$ (multiple) - Patterns match AST structure, not text - whitespace/formatting doesn’t matter

The Colon Issue: - Don’t include : in patterns - it’s part of Python’s concrete syntax, not the AST structure - ✅ def $FUNC($$$) - matches function definitions - ❌ def $FUNC($$$): - too specific, looking for the colon token itself

When to use kind vs pattern: - pattern: Simple direct matches (await $EXPR) - kind: Structural node types (kind: function_declaration)

Critical rule for relational searches: Always add stopBy: end to has/inside rules to search the entire subtree:

has:
  pattern: await $EXPR
  stopBy: end

Escaping in shell: Use \$VAR or single quotes when using --inline-rules from command line

_ast_id = await add_msg("print('hello')\nprint('world')\nlog('keep')", msg_type='code')

print(await msg_ast_replace(_ast_id, 'print($A)', 'logger.info($A)'))

@@ -1,3 +1,3 @@
-print('hello')
-print('world')
+logger.info('hello')
+logger.info('world')
 log('keep')

print((await read_msg(n=0, id=_ast_id, nums=True))['content'])

     1 │ logger.info('hello')
     2 │ logger.info('world')
     3 │ log('keep')

await del_msgs(_ast_id)

['6ebebb38']

Context

---

source

ctx_folder

async def ctx_folder(
    path:pathlib.Path='.', # Path to collect
    types:str | list='py,doc', # list or comma-separated str of ext types from: py, js, java, c, cpp, rb, r, ex, sh, web, doc, cfg
    out:bool=False, # Include notebook cell outputs?
    raw:bool=True, # Add raw message, or note?
    exts:str | list=None, # list or comma-separated str of exts to include (overrides `types`)
    *, prefix:bool=False, # Include Anthropic's suggested prose intro?
    include_base:bool=True, # Include full path in src?
    title:str=None, # Optional title attr for Documents element
    max_size:int=100000, # Skip files larger than this (bytes)
    max_total:int=10000000, # Max total output size in bytes
    readme_first:bool=False, # Prioritize README files at start of context?
    files_only:bool=False, # Return dict of {filename: size} instead of context?
    sigs_only:bool=False, # Return signatures instead of full text? (where supported by `codesigs` lib)
    ids:bool=True, # Include cell ids in notebooks?
    recursive:bool=True, # search subfolders
    maxdepth:int=None, # max depth to descend (1=just immediate contents; None=unlimited)
    symlinks:bool=True, # follow symlinks?
    file_glob:str=None, # Only include files matching glob
    file_re:str=None, # Only include files matching regex
    folder_re:str=None, # Only enter folders matching regex
    skip_file_glob:str=None, # Skip files matching glob
    skip_file_re:str=None, # Skip files matching regex
    skip_folder_re:str=None, # Skip folders matching regex,
    ret_folders:bool=False, # return folders, not just files
    sort:bool=True, # sort files by name within each folder
):

Convert folder to XML context and place in a new message

# ctx_folder('..', max_total=600, sigs_only=True, exts='py')

../dialoghelper/capture.py def setup_share(): “Setup screen sharing”

def start_share(): fire_event(‘shareScreen’)

def _capture_screen(timeout=15):

def capture_screen(timeout=15): “Capture the screen as a PIL image.”

def capture_tool(timeout:int=15): “Capture the screen. Re-call this function to get the most recent screenshot, as needed. Use default timeout where possible” ../dialoghelper/core.py <

[TRUNCATED: output size 24344 exceeded max size 600 bytes]

---

source

ctx_repo

async def ctx_repo(
    owner:str, # GitHub repo owner
    repo:str, # GitHub repo name
    types:str | list='py,doc', # list or comma-separated str of ext types from: py, js, java, c, cpp, rb, r, ex, sh, web, doc, cfg
    exts:str | list=None, # list or comma-separated str of exts to include (overrides `types`)
    out:bool=False, # Include notebook cell outputs?
    raw:bool=True, # Add raw message, or note?
    *, ref:str=None, # Git ref (branch/tag/sha) (get from URL not provided); defaults to repo's default branch
    folder:str=None, # Only include files under this path (get from URL not provided)
    show_filters:bool=True, # Include filter info in title?
    token:str=None, # GitHub token (uses GITHUB_TOKEN env var if None)
    prefix:bool=False, # Include Anthropic's suggested prose intro?
    max_size:int=100000, # Skip files larger than this (bytes)
    max_total:int=10000000, # Max total output size in bytes
    files_only:bool=False, # Return dict of {filename: size} instead of context?
    sigs_only:bool=False, # Return signatures instead of full text? (where supported by `codesigs` lib)
    ids:bool=True, # Include cell ids in notebooks?
    recursive:bool=True, # search subfolders
    maxdepth:int=None, # max depth to descend (1=just immediate contents; None=unlimited)
    symlinks:bool=True, # follow symlinks?
    file_glob:str=None, # Only include files matching glob
    file_re:str=None, # Only include files matching regex
    folder_re:str=None, # Only enter folders matching regex
    skip_file_glob:str=None, # Skip files matching glob
    skip_file_re:str=None, # Skip files matching regex
    skip_folder_re:str=None, # Skip folders matching regex,
    ret_folders:bool=False, # return folders, not just files
    sort:bool=True, # sort files by name within each folder
): # XML for LM context, or dict of file sizes

Convert GitHub repo to XML context and place in a new message

---

source

ctx_symfile

async def ctx_symfile(
    sym
):

Add note with filepath and contents for a symbol’s source file

# ctx_symfile(TemporaryDirectory)

---

source

ctx_symfolder

async def ctx_symfolder(
    sym, # Symbol to get folder context from
    *,
    types:str | list='py', # List or comma-separated str of ext types from: py, js, java, c, cpp, rb, r, ex, sh, web, doc, cfg
    skip_file_re:str='^_mod', # Skip files matching regex
    path:Union[str, pathlib.Path]='.', # Folder to read
    prefix:bool=False, # Include Anthropic's suggested prose intro?
    out:bool=True, # Include notebook cell outputs?
    include_base:bool=True, # Include full path in src?
    title:str=None, # Optional title attr for Documents element
    max_size:int=100000, # Skip files larger than this (bytes)
    max_total:int=10000000, # Max total output size in bytes
    readme_first:bool=False, # Prioritize README files at start of context?
    files_only:bool=False, # Return dict of {filename: size} instead of context?
    sigs_only:bool=False, # Return signatures instead of full text? (where supported by `codesigs` lib)
    ids:bool=True, # Include cell ids in notebooks?
    recursive:bool=True, # search subfolders
    maxdepth:int=None, # max depth to descend (1=just immediate contents; None=unlimited)
    symlinks:bool=True, # follow symlinks?
    file_glob:str=None, # Only include files matching glob
    file_re:str=None, # Only include files matching regex
    folder_re:str=None, # Only enter folders matching regex
    skip_file_glob:str=None, # Skip files matching glob
    skip_folder_re:str=None, # Skip folders matching regex,
    ret_folders:bool=False, # return folders, not just files
    sort:bool=True, # sort files by name within each folder
    exts:str | list=None, # list or comma-separated str of exts to include
):

Add raw message with folder context for a symbol’s source file location

# ctx_symfolder(folder2ctx)

---

source

ctx_sympkg

async def ctx_sympkg(
    sym, # Symbol to get folder context from
    *,
    types:str | list='py', # List or comma-separated str of ext types from: py, js, java, c, cpp, rb, r, ex, sh, web, doc, cfg
    skip_file_re:str='^_mod', # Skip files matching regex
    path:Union[str, pathlib.Path]='.', # Folder to read
    prefix:bool=False, # Include Anthropic's suggested prose intro?
    out:bool=True, # Include notebook cell outputs?
    include_base:bool=True, # Include full path in src?
    title:str=None, # Optional title attr for Documents element
    max_size:int=100000, # Skip files larger than this (bytes)
    max_total:int=10000000, # Max total output size in bytes
    readme_first:bool=False, # Prioritize README files at start of context?
    files_only:bool=False, # Return dict of {filename: size} instead of context?
    sigs_only:bool=False, # Return signatures instead of full text? (where supported by `codesigs` lib)
    ids:bool=True, # Include cell ids in notebooks?
    recursive:bool=True, # search subfolders
    maxdepth:int=None, # max depth to descend (1=just immediate contents; None=unlimited)
    symlinks:bool=True, # follow symlinks?
    file_glob:str=None, # Only include files matching glob
    file_re:str=None, # Only include files matching regex
    folder_re:str=None, # Only enter folders matching regex
    skip_file_glob:str=None, # Skip files matching glob
    skip_folder_re:str=None, # Skip folders matching regex,
    ret_folders:bool=False, # return folders, not just files
    sort:bool=True, # sort files by name within each folder
    exts:str | list=None, # list or comma-separated str of exts to include
):

Add raw message with repo context for a symbol’s root package

# ctx_sympkg(folder2ctx)

Gists and github

---

source

import_string

def import_string(
    code:str, # Code to import as a module
    name:str, # Name of module to create
):

def hi(who:str):
    "Say hi to `who`"
    return f"Hello {who}"

def hi2(who):
    "Say hi to `who`"
    return f"Hello {who}"

def hi3(who:str):
    return f"Hello {who}"

bye = "bye"

assert is_usable_tool(hi)
assert not is_usable_tool(hi2)
assert not is_usable_tool(hi3)
assert not is_usable_tool(bye)

---

source

mk_toollist

def mk_toollist(
    syms
):

print(mk_toollist([hi]))

- &`hi`: Say hi to `who`

---

source

import_gist

def import_gist(
    gist_id:str, # user/id or just id of gist to import as a module
    mod_name:str=None, # module name to create (taken from gist filename if not passed)
    add_global:bool=True, # add module to caller's globals?
    import_wildcard:bool=False, # import all exported symbols to caller's globals
    create_msg:bool=False, # Add a message that lists usable tools
):

Import gist directly from string without saving to disk

gistid = 'jph00/e7cfd4ded593e8ef6217e78a0131960c'
import_gist(gistid)
importtest.testfoo

---------------------------------------------------------------------------
KeyboardInterrupt                         Traceback (most recent call last)
Cell In[75], line 2
      1 gistid = 'jph00/e7cfd4ded593e8ef6217e78a0131960c'
----> 2 import_gist(gistid)
      3 importtest.testfoo

Cell In[73], line 10, in import_gist(gist_id, mod_name, add_global, import_wildcard, create_msg)
      6     import_wildcard:bool=False, # import all exported symbols to caller's globals
      7     create_msg:bool=False # Add a message that lists usable tools
      8 ):
      9     "Import gist directly from string without saving to disk"
---> 10     fil = GhApi(sync=True).gist_file(gist_id)
     11     mod_name = mod_name or Path(fil['filename']).stem
     12     module = import_string(fil['content'], mod_name)
     13     glbs = currentframe().f_back.f_globals

File ~/aai-ws/ghapi/ghapi/core.py:224, in gist_file(self, gist_id)
    221 @patch
    222 def gist_file(self:GhApi, gist_id:str):
    223     "Get the first file from a gist; coro if async client"
--> 224     return then(self.load_gist(gist_id), ~Self.files.values(), first)

File ~/aai-ws/ghapi/ghapi/core.py:219, in load_gist(self, gist_id)
    217 if '/' in gist_id: *_,user,gist_id = gist_id.split('/')
    218 else: user = None
--> 219 return self.gists.get(gist_id, user=user)

File ~/aai-ws/fastspec/fastspec/oapi.py:157, in SyncOpFunc.__call__(self, *args, **kwargs)
    155 if stream: raise TypeError("stream=True needs an async client; or wrap the async client with `fastcore.aio.iter_sync`")
    156 body = kw.pop('body')
--> 157 try: return dict2obj(self.client.request(self.verb, url, headers=headers, params=query, json=body, **kw))
    158 except Exception as e: self._raise_with_context(e, endpoint='', route=route, query=query, body=body)

File ~/aai-ws/ghapi/ghapi/core.py:90, in GhSyncTransport.request(self, method, url, raw, **kwargs)
     88 def request(self, method, url, *, raw=False, **kwargs):
     89     self._pre(method, url, kwargs)
---> 90     return self._post(SyncTransport.request(self, method, url, raw=True, **kwargs), raw)

File ~/aai-ws/fasttransport/fasttransport/core.py:105, in SyncTransport.request(self, method, url, headers, params, json, data, files, content, raw)
    103 "Sync version of `AsyncTransport.request`."
    104 with self._client() as client:
--> 105     resp = client.request(method, url, headers=self._request_headers(headers, files=files),
    106         params=params, json=json, data=data, files=files, content=content)
    107     try: resp.raise_for_status()
    108     except httpx2.HTTPStatusError as e:

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpx2/_client.py:797, in Client.request(self, method, url, content, data, files, json, params, headers, cookies, auth, follow_redirects, timeout, extensions)
    782     warnings.warn(message, DeprecationWarning, stacklevel=2)
    784 request = self.build_request(
    785     method=method,
    786     url=url,
   (...)    795     extensions=extensions,
    796 )
--> 797 return self.send(request, auth=auth, follow_redirects=follow_redirects)

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpx2/_client.py:980, in Client.send(self, request, stream, auth, follow_redirects)
    976 self._set_timeout(request)
    978 auth = self._build_request_auth(request, auth)
--> 980 response = self._send_handling_auth(
    981     request,
    982     auth=auth,
    983     follow_redirects=follow_redirects,
    984     history=[],
    985 )
    986 try:
    987     if not stream:

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpx2/_client.py:1008, in Client._send_handling_auth(self, request, auth, follow_redirects, history)
   1005 request = next(auth_flow)
   1007 while True:
-> 1008     response = self._send_handling_redirects(
   1009         request,
   1010         follow_redirects=follow_redirects,
   1011         history=history,
   1012     )
   1013     try:
   1014         try:

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpx2/_client.py:1043, in Client._send_handling_redirects(self, request, follow_redirects, history)
   1040 for hook in self._event_hooks["request"]:
   1041     hook(request)
-> 1043 response = self._send_single_request(request)
   1044 try:
   1045     for hook in self._event_hooks["response"]:

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpx2/_client.py:1076, in Client._send_single_request(self, request)
   1073     raise RuntimeError("Attempted to send an async request with a sync Client instance.")
   1075 with request_context(request=request):
-> 1076     response = transport.handle_request(request)
   1078 assert isinstance(response.stream, SyncByteStream)
   1080 response.request = request

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpx2/_transports/default.py:245, in HTTPTransport.handle_request(self, request)
    232 req = httpcore2.Request(
    233     method=request.method,
    234     url=httpcore2.URL(
   (...)    242     extensions=request.extensions,
    243 )
    244 with map_httpcore_exceptions():
--> 245     resp = self._pool.handle_request(req)
    247 assert isinstance(resp.stream, typing.Iterable)
    249 return Response(
    250     status_code=resp.status,
    251     headers=resp.headers,
    252     stream=ResponseStream(resp.stream),
    253     extensions=resp.extensions,
    254 )

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_sync/connection_pool.py:242, in ConnectionPool.handle_request(self, request)
    239         closing = self._assign_requests_to_connections()
    241     self._close_connections(closing)
--> 242     raise exc from None
    244 # Return the response. Note that in this case we still have to manage
    245 # the point at which the response is closed.
    246 assert isinstance(response.stream, typing.Iterable)

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_sync/connection_pool.py:224, in ConnectionPool.handle_request(self, request)
    220 connection = pool_request.wait_for_connection(timeout=timeout)
    222 try:
    223     # Send the request on the assigned connection.
--> 224     response = connection.handle_request(pool_request.request)
    225 except ConnectionNotAvailable:
    226     # In some cases a connection may initially be available to
    227     # handle a request, but then become unavailable.
    228     #
    229     # In this case we clear the connection and try again.
    230     pool_request.clear_connection()

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_sync/connection.py:96, in HTTPConnection.handle_request(self, request)
     93     self._connect_failed = True
     94     raise exc
---> 96 return self._connection.handle_request(request)

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_sync/http11.py:125, in HTTP11Connection.handle_request(self, request)
    123     with Trace("response_closed", logger, request) as trace:
    124         self._response_closed()
--> 125 raise exc

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_sync/http11.py:97, in HTTP11Connection.handle_request(self, request)
     88     pass
     90 with Trace("receive_response_headers", logger, request, kwargs) as trace:
     91     (
     92         http_version,
     93         status,
     94         reason_phrase,
     95         headers,
     96         trailing_data,
---> 97     ) = self._receive_response_headers(**kwargs)
     98     trace.return_value = (
     99         http_version,
    100         status,
    101         reason_phrase,
    102         headers,
    103     )
    105 network_stream = self._network_stream

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_sync/http11.py:167, in HTTP11Connection._receive_response_headers(self, request)
    164 timeout = timeouts.get("read", None)
    166 while True:
--> 167     event = self._receive_event(timeout=timeout)
    168     if isinstance(event, h11.Response):
    169         break

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_sync/http11.py:200, in HTTP11Connection._receive_event(self, timeout)
    197     event = self._h11_state.next_event()
    199 if event is h11.NEED_DATA:
--> 200     data = self._network_stream.read(self.READ_NUM_BYTES, timeout=timeout)
    202     # If we feed this case through h11 we'll raise an exception like:
    203     #
    204     #     httpcore2.RemoteProtocolError: can't handle event type
   (...)    208     # perspective. Instead we handle this case distinctly and treat
    209     # it as a ConnectError.
    210     if data == b"" and self._h11_state.their_state == h11.SEND_RESPONSE:

File ~/aai-ws/.venv/lib/python3.13/site-packages/httpcore2/_backends/sync.py:127, in SyncStream.read(self, max_bytes, timeout)
    125 with map_exceptions(exc_map):
    126     self._sock.settimeout(timeout)
--> 127     return self._sock.recv(max_bytes)

File ~/.local/share/uv/python/cpython-3.13.15-macos-aarch64-none/lib/python3.13/ssl.py:1297, in SSLSocket.recv(self, buflen, flags)
   1293     if flags != 0:
   1294         raise ValueError(
   1295             "non-zero flags not allowed in calls to recv() on %s" %
   1296             self.__class__)
-> 1297     return self.read(buflen)
   1298 else:
   1299     return super().recv(buflen, flags)

File ~/.local/share/uv/python/cpython-3.13.15-macos-aarch64-none/lib/python3.13/ssl.py:1152, in SSLSocket.read(self, len, buffer)
   1150         return self._sslobj.read(len, buffer)
   1151     else:
-> 1152         return self._sslobj.read(len)
   1153 except SSLError as x:
   1154     if x.args[0] == SSL_ERROR_EOF and self.suppress_ragged_eofs:

File ~/aai-ws/kernmini/kernmini/kernel.py:821, in MiniKernel.handle_sigint(self, signum, frame)
    819 if parent.cancel_async_execution(wake=True): return
    820 if not parent.sync_executing.is_set(): return
--> 821 raise KeyboardInterrupt

KeyboardInterrupt: 

import_gist.__doc__

import_gist(gistid, import_wildcard=True)
importtest.testfoo

hi("Sarah")

importtest.__all__

Input

input can take a string prompt as normal. OR, you can supply custom UI. For it to work, you must:

• Wrap in <solveit-input> tags (done automatically by the new input function defined here)
• Post to "/input_reply_" with a value for user_input.
• For access keys (shortcuts for buttons), include an accesskey="y" in the button, and make sure it is in a form/div with id 'input-request-form'.

The show_prompt demo here (using the custom InputBtn) demonstrates this to give a custom yes/no prompt:

---

source

InputForm

def InputForm(
    *c, **kwargs
):

Create an input() with a Form with needed hx_post and id

---

source

input

def input(
    prompt:str='', *args
):

Solveit customised input to handle fasttag prompts

---

source

InputBtn

def InputBtn(
    txt, value:NoneType=None, btncls:tuple=(), **kw
):

def show_prompt():
    return InputForm(
        Div('Ship this change now?'),
        Div(cls='flex gap-2')(
            InputBtn('Yes', btncls='primary', accesskey="y"),
            InputBtn('No', btncls='default', accesskey="n"))
    )

# show_prompt()