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)

Use ast_grep to find calls to an API before changing it. Patterns match multiline code and nested expressions.

ast_grep calls the ast-grep CLI to search code by syntax tree. Matches ignore formatting differences.

The pattern xpost($A, data=$B, $$$) matches calls to xpost with:

• Any first argument, captured as $A.
• A data keyword argument, captured as $B.
• Zero or more trailing arguments, matched by $$$.

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')]

A pattern matches code, such as await $EXPR or console.log($ARG). A kind rule matches a syntax tree node type.

To search the entire subtree with has or inside, set stopBy: end:

has:
  pattern: await $EXPR
  stopBy: end

In shell commands, use single quotes around patterns or escape $ as \$ to prevent variable expansion.

_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
    path_glob:str=None, # Only include files whose full paths match glob (wildcards match separators)
    path_re:str=None, # Only include files whose full paths match 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
    path_glob:str=None, # Only include files whose full paths match glob (wildcards match separators)
    path_re:str=None, # Only include files whose full paths match 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
    path_glob:str=None, # Only include files whose full paths match glob (wildcards match separators)
    path_re:str=None, # Only include files whose full paths match 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
    path_glob:str=None, # Only include files whose full paths match glob (wildcards match separators)
    path_re:str=None, # Only include files whose full paths match 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 accepts a string prompt or custom UI. Custom UI needs:

• A <solveit-input> wrapper, which input adds automatically.
• A POST to /input_reply_ with a value for user_input.
• For keyboard shortcuts, an accesskey on the button and a containing form or div with ID input-request-form.

show_prompt uses InputBtn to provide a 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()