examples/enact_concepts.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Enact Concepts\n",
    "\n",
    "\n",
    "\n",
    "## Resources\n",
    "\n",
    "Enact is a framework for generative software, which offers the ability to track\n",
    "python programs and their executions. To this end, any tracked object in the\n",
    "enact framework is an instance of a `Resource`, including data, executable\n",
    "components and journaled executions.\n",
    "\n",
    "New resources can be defined and registered with the enact framework as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import enact\n",
    "\n",
    "import dataclasses\n",
    "\n",
    "@enact.register\n",
    "@dataclasses.dataclass\n",
    "class MyResource(enact.Resource):\n",
    "  my_field: str\n",
    "  my_other_field: int"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "A `Resource` subclass defines a unique type ID and an interface to access field\n",
    "names and their values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "TypeID of MyResource: '{\"digest\": \"a46cd7bf341ef134fe8fd7e97e564f0f84425c747c3c55850cd599f1249b97f5\", \"name\": \"__main__.MyResource\"}'\n",
      "Fields of my_resource: ['my_field', 'my_other_field']\n",
      "Value of my_field: hello\n",
      "Value of my_other_field: 42\n",
      "Field values of the resource: ['hello', 42]\n",
      "Field items: [('my_field', 'hello'), ('my_other_field', 42)]\n"
     ]
    }
   ],
   "source": [
    "print(f'TypeID of MyResource: {repr(MyResource.type_id())}')\n",
    "print(f'Fields of my_resource: {list(MyResource.field_names())}')\n",
    "\n",
    "my_resource = MyResource('hello', 42)\n",
    "print(f'Value of my_field: {my_resource.my_field}')\n",
    "print(f'Value of my_other_field: {my_resource.my_other_field}')\n",
    "print(f'Field values of the resource: {list(my_resource.field_values())}')\n",
    "print(f'Field items: {list(my_resource.field_items())}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Resources may be nested and they can contain maps and lists, but the\n",
    "leaf types are limited to:\n",
    "* Python primitives of type `int`, `float`, `bool`, `str`, `bytes`, `bool`,\n",
    "* the `None` value,\n",
    "* other resources (i.e., instances of `ResourceBase`),\n",
    "* types of resources (i.e., subclasses of `ResourceBase`).\n",
    "\n",
    "Python maps and lists are allowed and may be nested, but map keys must be of type `str`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'primitives': [1, 2.0, False, None, 'test', b'bytes'],\n",
       " 'map_value': {'hello': {'my_field': 'hello', 'my_other_field': 69}},\n",
       " 'nested_resource': {'my_field': 'hola', 'my_other_field': 42},\n",
       " 'resource_type': __main__.MyResource}"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from typing import Dict, List, Union, Type\n",
    "\n",
    "@dataclasses.dataclass\n",
    "class ComplexResource(enact.Resource):\n",
    "  primitives: List[Union[int, float, bool, str, bytes, bool, type(None)]]\n",
    "  map_value: Dict[str, MyResource]\n",
    "  nested_resource: MyResource\n",
    "  resource_type: Type[MyResource]\n",
    "\n",
    "ComplexResource(\n",
    "  primitives=[1, 2.0, False, None, 'test', b'bytes'],\n",
    "  map_value={'hello': MyResource('hello', 69)},\n",
    "  nested_resource=MyResource('hola', 42),\n",
    "  resource_type=MyResource).to_resource_dict()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Stores and References\n",
    "\n",
    "`Resource` instances can be committed to a store which yields an\n",
    "immutable reference to the instance in its current state. References\n",
    "can be checked out of the store."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<Ref: 947051f773d0e01ed76be7506fe4cfde5be54d35eef91f10b488d3e8b789ee3b>\n",
      "MyResource(my_field='hello', my_other_field=42)\n"
     ]
    }
   ],
   "source": [
    "with enact.InMemoryStore() as store:\n",
    "  # Enter the context of the store.\n",
    "  my_resource = MyResource('hello', 42)\n",
    "  ref = enact.commit(my_resource)  # commits the resource to the active store.\n",
    "  print(ref)\n",
    "  print(ref.checkout())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Since checking out references is a common operation, `ref.checkout()`\n",
    "may be abbreviated to simply `ref()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "MyResource(my_field='hello', my_other_field=42)\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  print(ref())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "References point to resources, but they are also resources themselves. This\n",
    "means that resources can have fields that reference other resources."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "MyResource(my_field='hola', my_other_field=69)\n",
      "MyResource(my_field='hello', my_other_field=42)\n"
     ]
    }
   ],
   "source": [
    "from typing import Optional\n",
    "\n",
    "\n",
    "@enact.register\n",
    "@dataclasses.dataclass\n",
    "class LinkedList(enact.Resource):\n",
    "  \"\"\"A linked list of resources.\"\"\"\n",
    "  value: enact.FieldValue\n",
    "  next: Optional[enact.Ref['LinkedList']] = None\n",
    "\n",
    "with store:\n",
    "  # Create a linked list with two nodes.\n",
    "  l = enact.commit(LinkedList(value=my_resource))\n",
    "  l = enact.commit(LinkedList(value=MyResource('hola', 69), next=l))\n",
    "  # Iterate through the list.\n",
    "  while l != None:\n",
    "    print(l().value)\n",
    "    l = l().next"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Invokables\n",
    "\n",
    "Invokables are executable resources that allow for journaled execution which can\n",
    "be rewound and replayed. They accept a resource-typed argument and produce a\n",
    "resource-typed output. Invokables may be annotated with their specific input and\n",
    "output types using the `typed_invokable` decorator, which will also register the\n",
    "resource with the enact framework."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "import random\n",
    "\n",
    "\n",
    "@enact.typed_invokable(enact.NoneResource, enact.Int)\n",
    "@dataclasses.dataclass\n",
    "class RollDie(enact.Invokable[enact.NoneResource, enact.Int]):\n",
    "  sides: int = 6\n",
    "  def call(self) -> enact.Int:\n",
    "    return enact.Int(random.randint(1, self.sides))\n",
    "\n",
    "@enact.typed_invokable(enact.Int, enact.Int)\n",
    "@dataclasses.dataclass\n",
    "class RollDice(enact.Invokable):\n",
    "  die: enact.Ref[enact.Invokable[enact.NoneResource, enact.Int]]\n",
    "  def call(self, num_rolls: enact.Int) -> enact.Int:\n",
    "    die = self.die.checkout()\n",
    "    return enact.Int(sum(die() for _ in range(num_rolls)))\n",
    "  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Invokables can be called just like any python callable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "9\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  # Create the die to use.\n",
    "  die = enact.commit(RollDie())\n",
    "  # Create invokable that rolls the die repeatedly.\n",
    "  dice = RollDice(die)\n",
    "  # Roll the die thrice and report sum.\n",
    "  print(dice(enact.Int(3)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Additionally they support journaled execution using the `invoke` function, which\n",
    "recursively tracks inputs, outputs and raised exceptions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Individual dice Rolls:\n",
      "  roll 0: 1\n",
      "  roll 1: 6\n",
      "  roll 2: 4\n",
      "Output is: -> Int(value=11)#2c847a\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  num_rolls = enact.commit(enact.Int(3))\n",
    "  invocation = dice.invoke(num_rolls)\n",
    "  \n",
    "def print_rolls(invocation):\n",
    "  print(f'Individual dice Rolls:')\n",
    "  for i, child in enumerate(invocation.get_children()):\n",
    "    print(f'  roll {i}: {child.get_output()}')\n",
    "  print(f'Output is: {enact.pformat(invocation.response().output)}')\n",
    "  \n",
    "print_rolls(invocation)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Advanced concepts\n",
    "\n",
    "### Storage model\n",
    "\n",
    "When committing a resource, the returned reference refers to a cryptographic\n",
    "digest of the resource type and it's content. \n",
    "\n",
    "This means that when committing an identical resource instance twice, the same\n",
    "reference is returned:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "True\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  r1 = enact.commit(MyResource('hello', 42))\n",
    "  r2 = enact.commit(MyResource('hello', 42))\n",
    "  print(r1 == r2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Conceptually, stores can be viewed as hashtables which, due to their use of long\n",
    "hash digests, have very low collision probability. This means that separate enact\n",
    "stores containing different data, can be interpreted as providing partial views \n",
    "on the same global 'address space'.\n",
    "\n",
    "Stores contain resources that may contain references to other resources. They\n",
    "can therefore be interpreted as directed graphs. For example, invocations\n",
    "point at the input and output resources.\n",
    "\n",
    "Note that it is not possible to commit two resources that reference each other:\n",
    "Taking a reference requires creating a cryptographic digest of resource\n",
    "contents, including all of its own reference fields. Therefore, to compute the\n",
    "digest of the first resource, we would need to first compute the digest of the\n",
    "second - but this would in turn require computing the digest of the first:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "-> LinkedList#0b182f:\n",
      "  value: 1\n",
      "  next:\n",
      "    -> LinkedList#0f9059:\n",
      "      value: 2\n",
      "      next: -> LinkedList(value=1, next=None)#dd2ab0\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  l1 = LinkedList(value=1, next=None)\n",
    "  l2 = LinkedList(value=2, next=enact.commit(l1))\n",
    "\n",
    "  # This does not create a cycle, since l2 points at the previously committed\n",
    "  # version of l1.\n",
    "  l1.next = enact.commit(l2)\n",
    "\n",
    "  # We now have list of length 3 instead of a cycle:\n",
    "  enact.pprint(enact.commit(l1))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notice that the digest (after the `#` symbol) is different for the three linked\n",
    "list nodes.\n",
    "\n",
    "Since resources can't reference each other store graphs are acyclic, which means\n",
    "that they form a _directed acyclic graph_ (DAG). DAGs that employ this form of\n",
    "hashing are known as Merkle DAGs or hash DAGs. Hash DAGs are generalizations of\n",
    "data structures such as Merkle trees and block chains."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Rewinding and replaying invocations\n",
    "\n",
    "Invocations represent journaled executions, that store the\n",
    "inputs and outputs of all intermediate invokables. Because of this,\n",
    "we can use them to replay a previous execution. For example, in order\n",
    "to reroll the last die in an dice roll execution we can call `rewind`\n",
    "to remove a specified number of calls from the invocation object and then\n",
    "`replay` the invocation, which will only reexecute the now missing calls."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Individual dice Rolls:\n",
      "  roll 0: 3\n",
      "  roll 1: 5\n",
      "  roll 2: 1\n",
      "Output is: -> Int(value=9)#c9c8e5\n",
      "\n",
      "==Rewound invocation to:\n",
      "Individual dice Rolls:\n",
      "  roll 0: 3\n",
      "Output is: None\n",
      "\n",
      "==Replayed invocation as:\n",
      "Individual dice Rolls:\n",
      "  roll 0: 3\n",
      "  roll 1: 3\n",
      "  roll 2: 5\n",
      "Output is: -> Int(value=11)#2c847a\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  invocation = dice.invoke(enact.commit(enact.Int(3)))\n",
    "  print_rolls(invocation)\n",
    "\n",
    "  invocation = invocation.rewind(2)\n",
    "  print('\\n==Rewound invocation to:')\n",
    "  print_rolls(invocation)\n",
    "\n",
    "  print('\\n==Replayed invocation as:')\n",
    "  invocation = invocation.replay()\n",
    "  print_rolls(invocation)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Note that the `call` function of any invocation that raised an exception will\n",
    "be reexecuted, although if it had subinvocations that succeeded, those will\n",
    "be replayed.\n",
    "\n",
    "Replays also support an `exception_override` argument which allows injecting\n",
    "values instead of replaying an exception. This is the mechanism with which\n",
    "[input requests](#input-requests) are handled:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Output: Injected value\n"
     ]
    }
   ],
   "source": [
    "class Halt(enact.ExceptionResource):\\\n",
    "  pass\n",
    "\n",
    "@enact.typed_invokable(enact.NoneResource, enact.Str)\n",
    "class RaiseHalt(enact.Invokable):\n",
    "  def call(self):\n",
    "    raise Halt()\n",
    "\n",
    "def exception_override(exception_ref):\n",
    "  if isinstance(exception_ref(), Halt):\n",
    "    return enact.Str('Injected value')\n",
    "\n",
    "with store:\n",
    "  invocation = RaiseHalt().invoke()\n",
    "  assert isinstance(invocation.get_raised(), Halt)\n",
    "  print(f'Output: {invocation.replay(exception_override).get_output()}')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Side note: Replays and non-determinism"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Replays work by reexecuting invokables that have no known output in the \n",
    "replayed invocation. During reexecution, calls to child invokables must \n",
    "be matched to the recorded calls in order to see if an output can be replayed.\n",
    "Therefore, for the replay functionality to work properly, the next call to a\n",
    "subinvokable must be a deterministic function of the invokable input and the\n",
    "return values of previous subinvokables.\n",
    "\n",
    "That is, all sources of non-determinism must either be wrapped in invokable\n",
    "objects or must make no difference to the call sequence. For example, for\n",
    "subinvokables `x`, `y`, `z`:\n",
    "1. `return random.choice([y, z])()`: Not replayable, since non-determinism\n",
    "affects the call order.\n",
    "2. `return x() ? y() : z()`: Replayable, since any non-determinism in `x` is\n",
    "contained within an invokable.\n",
    "3. `return enact.Int(x() + random.randint(0, 5))`: Replayable, since\n",
    "non-determinism does not effect call order or call arguments to subinvokables.\n",
    "\n",
    "The following example shows what happens when this property is violated."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Got error: Expected invocation FormatTimestamp()(1692134680.0698006) but got FormatTimestamp()(1692134680.067802).\n",
      "Ensure that calls to subinvokable are deterministic or use strict=False.\n"
     ]
    }
   ],
   "source": [
    "import datetime\n",
    "import time\n",
    "\n",
    "\n",
    "@enact.typed_invokable(enact.Float, enact.Str)\n",
    "class FormatTimestamp(enact.Invokable):\n",
    "  \"\"\"Formats a timestamp as a string.\"\"\"\n",
    "\n",
    "  def call(self, timestamp: enact.Float) -> enact.Str:\n",
    "    return enact.Str(datetime.datetime.fromtimestamp(timestamp))\n",
    "\n",
    "@enact.typed_invokable(enact.NoneResource, enact.Str)\n",
    "class FormatCurrentTime(enact.Invokable):\n",
    "  \"\"\"Return the current time as a string.\"\"\"\n",
    "  \n",
    "  def call(self) -> enact.Str:\n",
    "    now = enact.Float(time.time())\n",
    "    # ERROR: Non-deterministic call argument to subinvokable:\n",
    "    return FormatTimestamp()(now)\n",
    "\n",
    "with store:\n",
    "  invocation = FormatCurrentTime().invoke()\n",
    "  # Delete only the output to force reexecution.\n",
    "  invocation = invocation.rewind(num_calls=0)\n",
    "  try:\n",
    "    invocation.replay()\n",
    "  except enact.ReplayError as e:\n",
    "    print(f'Got error: {e}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To fix the above example, we could either wrap `time.time()` in an invokable\n",
    "`CurrentTime` or, alternatively, define `FormatTimestamp` as a normal python\n",
    "function instead of making it an invokable."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Input Requests\n",
    "\n",
    "Input requests are a mechanism for interrupting an ongoing execution in order to\n",
    "collect external input from a user or system. (This is conceptually\n",
    "similar to generators or continuations).\n",
    "\n",
    "Input requests can be generated by calling the `request_input` function, which\n",
    "takes as arguments a resource type that is requested, an optional resource for\n",
    "which input is requested and a context for the request (e.g., instructions to \n",
    "a user).\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "@enact.typed_invokable(enact.NoneResource, enact.Int)\n",
    "class DiceRollWithUserConfirmation(enact.Invokable):\n",
    "  def call(self, num_calls: enact.Int) -> enact.Int:\n",
    "    roll_die = RollDie()\n",
    "    total = 0\n",
    "    for i in range(num_calls):\n",
    "      while True:\n",
    "        score = roll_die()\n",
    "        if enact.request_input(\n",
    "            requested_type=enact.Str,\n",
    "            for_resource=score,\n",
    "            context=f'Should I reroll die #{i}?') == 'yes':\n",
    "          continue\n",
    "        break\n",
    "      total += score\n",
    "    return enact.Int(total)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The `InvocationGenerator` class provides a python `Generator` interface for\n",
    "processing an invocation step by step, stopping at InputRequests."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Total score: 17\n",
      "  Die roll: 5\n",
      "   -Rerolled? no\n",
      "  Die roll: 6\n",
      "   -Rerolled? no\n",
      "  Die roll: 3\n",
      "   -Rerolled? yes\n",
      "  Die roll: 4\n",
      "   -Rerolled? yes\n",
      "  Die roll: 1\n",
      "   -Rerolled? yes\n",
      "  Die roll: 1\n",
      "   -Rerolled? yes\n",
      "  Die roll: 2\n",
      "   -Rerolled? yes\n",
      "  Die roll: 1\n",
      "   -Rerolled? yes\n",
      "  Die roll: 3\n",
      "   -Rerolled? yes\n",
      "  Die roll: 3\n",
      "   -Rerolled? yes\n",
      "  Die roll: 6\n",
      "   -Rerolled? no\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  inv_gen = enact.InvocationGenerator(\n",
    "    DiceRollWithUserConfirmation(),\n",
    "    enact.commit(enact.Int(3)))\n",
    "  \n",
    "  for input_request in inv_gen:\n",
    "    roll = input_request.for_resource()\n",
    "    if roll < 5:\n",
    "      inv_gen.set_input(enact.Str('yes'))\n",
    "    else:\n",
    "      inv_gen.set_input(enact.Str('no'))\n",
    "  \n",
    "  # Print invocation history:\n",
    "  invocation = inv_gen.invocation\n",
    "  print('Total score:', invocation.get_output())\n",
    "  for child in invocation.get_children():\n",
    "    output = child.get_output()\n",
    "    if isinstance(output, enact.Int):\n",
    "      print('  Die roll:', output)\n",
    "    else:\n",
    "      print('   -Rerolled?', output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Modifying references"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "While references point to immutable snapshots of resource objects, it is\n",
    "possibly to change which resource a given `Ref` object points to using the\n",
    "`modify()` context manager:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ref points at new resource:\n",
      "  MyResource(my_field='hello', my_other_field=69)\n",
      "but the old resource is still intact:\n",
      "  MyResource(my_field='hello', my_other_field=42)\n"
     ]
    }
   ],
   "source": [
    "with store:\n",
    "  ref = enact.commit(MyResource('hello', 42))\n",
    "  ref_copy = ref.deep_copy_resource()\n",
    "  with ref.modify() as my_resource:\n",
    "    my_resource.my_other_field = 69\n",
    "  print(f'ref points at new resource:\\n  {ref()}')\n",
    "  print(f'but the old resource is still intact:\\n  {ref_copy()}')\n",
    "  "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Custom store backends and reference types\n",
    "\n",
    "A store is connected to a backend which defines how resources are actually\n",
    "stored. Calling `InMemoryStore()` is equivalent to calling\n",
    "`Store(InMemoryBackend())`, and similarly, `FileStore(root_dir)` is equivalent\n",
    "to `Store(FileBackend(root_dir))`.\n",
    "\n",
    "New storage backends can be created by implementing the following interface:\n",
    "\n",
    "```python\n",
    "class StorageBackend(abc.ABC):\n",
    "  \"\"\"A storage backend.\"\"\"\n",
    "\n",
    "  @abc.abstractmethod\n",
    "  def commit(self, packed_resource: PackedResource):\n",
    "    \"\"\"Stores a packed resource.\"\"\"\n",
    "\n",
    "  @abc.abstractmethod\n",
    "  def has(self, ref: Ref) -> bool:\n",
    "    \"\"\"Returns whether the storage backend has the resource.\"\"\"\n",
    "\n",
    "  @abc.abstractmethod\n",
    "  def get(self, ref: Ref) -> Optional[interfaces.ResourceDict]:\n",
    "    \"\"\"Returns the packed resource or None if not available.\"\"\"\n",
    "```\n",
    "\n",
    "Resources are _packed_ into serializable dictionaries before being committed.\n",
    "This additional step can be overriden in custom `Ref` subclasses in order to\n",
    "support advanced functionality such as compression or end-to-end encryption of\n",
    "resources.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Inheriting from ResourceBase\n",
    "\n",
    "Extending the `Resource` class is the most convenient way to define new\n",
    "resources, since they use the dataclass interface to implement various\n",
    "`ResourceBase` functions. In some cases, it can be preferable to directly\n",
    "implement the interface. An example is the case where one wants to wrap an \n",
    "existing type and expose its data using the `ResourceBase` interface."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<Ref: eba94b35ab9af87fd2e8746d5b5eed4af99ac2468541899ff976b1e70d70cc97>\n",
      "3\n"
     ]
    }
   ],
   "source": [
    "from typing import Iterable, Mapping\n",
    "\n",
    "from enact import interfaces\n",
    "\n",
    "@enact.register\n",
    "class MyInt(int, enact.ResourceBase):\n",
    "  @classmethod\n",
    "  def field_names(cls) -> Iterable[str]:\n",
    "    \"\"\"Returns the names of the fields of the resource.\"\"\"\n",
    "    return ('value',)\n",
    "\n",
    "  def field_values(self) -> Iterable[interfaces.FieldValue]:\n",
    "    \"\"\"Return a list of field values, aligned with field_names.\"\"\"\n",
    "    return (int(self),)\n",
    "\n",
    "  @classmethod\n",
    "  def from_fields(cls,\n",
    "                  field_dict: Mapping[str, interfaces.FieldValue]) -> 'MyInt':\n",
    "    \"\"\"Constructs the resource from a field dictionary.\"\"\"\n",
    "    value = field_dict['value']\n",
    "    return cls(value)  # type: ignore\n",
    "\n",
    "\n",
    "with store:\n",
    "  ref = enact.commit(MyInt(3))\n",
    "  print(ref)\n",
    "  print(ref.checkout())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Similarly, invokables can directly extend `InvokableBase` instead of using a\n",
    "dataclass-based `Invokable` implementation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Meta-invocations and higher-order functions.\n",
    "\n",
    "Invokables support two core operations:\n",
    "1. They can be _called_, like a normal python callable.\n",
    "2. They can be _invoked_, which corresponds to a journaled execution.\n",
    "\n",
    "Typically, invokables will _call_ other invokables, but they may also _invoke_\n",
    "them, which is useful if an invokable wants to study the execution path of\n",
    "another invokable.\n",
    "\n",
    "This is useful, for example, when dealing with AI-generated code, since the\n",
    "the execution trace of a generated function can be used as feedback to the\n",
    "generation process.\n",
    "\n",
    "We illustrate the concept of meta-invocation with a simple example using dice\n",
    "wherein we create an invokable that analyzes another invokable's execution to\n",
    "determine the number of execution steps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Rolls until prime: 1\n",
      "The prime that was rolled: 3\n"
     ]
    }
   ],
   "source": [
    "@enact.typed_invokable(RollDie, enact.Int)\n",
    "class RollUntilPrime(enact.Invokable):\n",
    "  \"\"\"Rolls a die until the sum of the rolls is prime.\"\"\"\n",
    "  def call(self, die: RollDie) -> enact.Int:\n",
    "    total = 0\n",
    "    def is_prime(n):\n",
    "      return n > 1 and all(n % i for i in range(2, n))\n",
    "    while not is_prime(total):\n",
    "      total += die()\n",
    "    return enact.Int(total)\n",
    "\n",
    "@enact.typed_invokable(enact.Request, enact.Invocation)\n",
    "class Invoke(enact.Invokable):\n",
    "  \"\"\"Returns the result of invoking the given request.\"\"\"\n",
    "  def call(self, request: enact.Request) -> enact.Invocation:\n",
    "    return request.invokable().invoke(request.input)\n",
    "\n",
    "@enact.typed_invokable(RollDie, enact.Int)\n",
    "class CountRolls(enact.Invokable):\n",
    "  \"\"\"Analyze the execution trace to see how long it takes to roll a prime.\"\"\"\n",
    "  def call(self, die: RollDie) -> enact.Int:\n",
    "    invoke = Invoke()\n",
    "    invocation = invoke(enact.Request(\n",
    "      enact.commit(RollUntilPrime()), enact.commit(die)))\n",
    "    return enact.Int(len(list(invocation.get_children())))\n",
    "\n",
    "with store:\n",
    "  count_rolls = CountRolls()\n",
    "  invocation = count_rolls.invoke(enact.commit(RollDie(6)))\n",
    "  print(f'Rolls until prime: {invocation.get_output()}')\n",
    "  # Get output of call to invoke(...)\n",
    "  meta_invocation = invocation.get_child(0).get_output()\n",
    "  print(f'The prime that was rolled: {meta_invocation.get_output()}')\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Support for asyncio\n",
    "\n",
    "Python coroutines provide a model for concurrent execution of python code. This\n",
    "is useful in cases where, e.g., a generative component needs to query multiple\n",
    "APIs. Instead of waiting on the output of one call before executing the next,\n",
    "both calls can be issued concurrently, e.g.:\n",
    "\n",
    "```python\n",
    "import asyncio\n",
    "\n",
    "# Define a coroutine using the 'async' keyword.\n",
    "async def wait_on_api_calls():\n",
    "  # Use the 'await' keyword to suspend execution until both the api_call1 and\n",
    "  # the api_call2 coroutines have completed.\n",
    "  result = await asyncio.gather(api_call1(), api_call2())\n",
    "  return result\n",
    "\n",
    "def main():\n",
    "  asyncio.run(wait_on_api_calls())  # Run the coroutine in an event loop.\n",
    "```\n",
    "\n",
    "For more information on `async`, see the [python documentation](https://docs.python.org/3.8/library/asyncio.html).\n",
    "\n",
    "Enact supports asyncio components through the `AsyncInvokable` base class. \n",
    "Working with `AsyncInvokable` components is similar to working with `Invokable`\n",
    "components, except that the `call` and `invoke` functions are both\n",
    "coroutines defined using the `async` keyword."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Rolling die nr 0.\n",
      "Rolling die nr 1.\n",
      "Rolling die nr 2.\n",
      "Got result for die nr 1: 6\n",
      "Got result for die nr 0: 3\n",
      "Got result for die nr 2: 4\n",
      "Sum: 13\n",
      "\n",
      "Replay last roll:\n",
      "Rolling die nr 2.\n",
      "Got result for die nr 2: 5\n",
      "Sum: 14\n"
     ]
    }
   ],
   "source": [
    "import asyncio\n",
    "\n",
    "@enact.typed_invokable(enact.NoneResource, enact.Int)\n",
    "@dataclasses.dataclass\n",
    "class AsyncDieRoll(enact.AsyncInvokable):\n",
    "  \"\"\"Asynchronously rolls a die and returns the result.\"\"\"\n",
    "  dice_nr: int\n",
    "\n",
    "  async def api_call(self) -> int:\n",
    "    \"\"\"Mimic calling into an API to obtain roll result.\"\"\"\n",
    "    print(f'Rolling die nr {self.dice_nr}.')\n",
    "    await asyncio.sleep(random.random() * 0.1)\n",
    "    result = random.randint(1, 6)\n",
    "    print(f'Got result for die nr {self.dice_nr}: {result}')\n",
    "    return result\n",
    "\n",
    "  async def call(self) -> enact.Int:\n",
    "    return enact.Int(await self.api_call())\n",
    "\n",
    "\n",
    "@enact.typed_invokable(enact.Int, enact.Int)\n",
    "class AsyncDiceRoll(enact.AsyncInvokable):\n",
    "  \"\"\"Asynchronously rolls the requested number of dice.\"\"\"\n",
    "\n",
    "  async def call(self, num_dice: enact.Int) -> enact.Int:\n",
    "    \"\"\"Roll the requested number of dice concurrently.\"\"\"\n",
    "    dice = [AsyncDieRoll(i) for i in range(num_dice)]\n",
    "    rolls = [die() for die in dice]\n",
    "    return enact.Int(sum(await asyncio.gather(*rolls)))\n",
    "\n",
    "with store:\n",
    "  # Jupyter notebooks have their own event loop, so we can use await rather than\n",
    "  # asyncio.run:\n",
    "  invocation = await AsyncDiceRoll().invoke(num_rolls)\n",
    "  print(f'Sum: {invocation.get_output()}')\n",
    "  \n",
    "  print(f'\\nReplay last roll:')\n",
    "  invocation = invocation.rewind()  # Rewind the last call.\n",
    "  invocation = await invocation.replay_async()\n",
    "  print(f'Sum: {invocation.get_output()}')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Python coroutines allow scheduling background tasks, which execute outside\n",
    "of their calling context. Since enact needs to track the results of\n",
    "subinvocations, this is not supported and will raise an error."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Rolling die nr 0.\n",
      "Rolling die nr 1.\n",
      "Rolling die nr 2.\n",
      "Got result for die nr 1: 6\n",
      "Got result for die nr 0: 5\n",
      "Invocation failed: Subinvocation 2 did not complete during invocation of parent: AsyncDieRoll(dice_nr=2) invoked on <enact.interfaces.NoneResource object at 0x0000027E7E572F70>\n"
     ]
    }
   ],
   "source": [
    "@enact.typed_invokable(enact.Int, enact.Int)\n",
    "class WaitForFirstResult(enact.AsyncInvokable):\n",
    "  \"\"\"Rolls the requested number of dice and waits on the first result.\"\"\"\n",
    "\n",
    "  async def call(self, num_dice: enact.Int) -> enact.Int:\n",
    "    \"\"\"Roll the requested number of dice concurrently.\"\"\"\n",
    "    loop = asyncio.get_running_loop()\n",
    "    # Create background tasks for each die roll.\n",
    "    tasks = [loop.create_task(AsyncDieRoll(i)()) for i in range(num_dice)]\n",
    "    # Wait for the first result.\n",
    "    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) \n",
    "    # Leave pending tasks running.\n",
    "    return await done[0]\n",
    "\n",
    "with store:\n",
    "  try:\n",
    "    invocation = await WaitForFirstResult().invoke(num_rolls)\n",
    "  except enact.IncompleteSubinvocationError as e:\n",
    "    print(f\"Invocation failed: {e}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Requesting and resolving multiple inputs.\n",
    "\n",
    "Note that `async` can be used to request multiple inputs from external sources at once:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "result: Traceback (most recent call last):\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 622, in async_call\n",
      "    output_resource = await ReplayContext.async_call_or_replay(\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 398, in async_call_or_replay\n",
      "    result = await call(arg)\n",
      "  File \"C:\\Users\\leo\\AppData\\Local\\Temp\\ipykernel_26484\\2242140787.py\", line 19, in call\n",
      "    raise r\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 783, in __call__\n",
      "    output = await builder.async_call()\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 622, in async_call\n",
      "    output_resource = await ReplayContext.async_call_or_replay(\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 398, in async_call_or_replay\n",
      "    result = await call(arg)\n",
      "  File \"C:\\Users\\leo\\AppData\\Local\\Temp\\ipykernel_26484\\2242140787.py\", line 8, in call\n",
      "    return enact.request_input(enact.Int, for_resource=enact.Int(self.roll_id))\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 889, in request_input\n",
      "    return RequestInput(requested_type, context)(for_resource)\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 722, in __call__\n",
      "    output = builder.call()\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 602, in call\n",
      "    output_resource = ReplayContext.call_or_replay(\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 370, in call_or_replay\n",
      "    replayed_output, child_ctx = context._consume_replay(invokable, arg)\n",
      "  File \"C:\\dev\\enact\\src\\enact\\invocations.py\", line 443, in _consume_replay\n",
      "    override = self._exception_override(replay_response.raised)\n",
      "  File \"C:\\Users\\leo\\AppData\\Local\\Temp\\ipykernel_26484\\2242140787.py\", line 32, in provide_input\n",
      "    print(f'injecting input {roll} for roll {input_request.input()}')\n",
      "AttributeError: 'InputRequest' object has no attribute 'input'\n",
      "\n"
     ]
    },
    {
     "ename": "AssertionError",
     "evalue": "",
     "output_type": "error",
     "traceback": [
      "\u001b[1;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[1;31mAssertionError\u001b[0m                            Traceback (most recent call last)",
      "Cell \u001b[1;32mIn[26], line 37\u001b[0m\n\u001b[0;32m     35\u001b[0m invocation \u001b[39m=\u001b[39m \u001b[39mawait\u001b[39;00m invocation\u001b[39m.\u001b[39mreplay_async(exception_override\u001b[39m=\u001b[39mprovide_input)\n\u001b[0;32m     36\u001b[0m \u001b[39mprint\u001b[39m(\u001b[39mf\u001b[39m\u001b[39m'\u001b[39m\u001b[39mresult: \u001b[39m\u001b[39m{\u001b[39;00minvocation\u001b[39m.\u001b[39mget_raised()\u001b[39m}\u001b[39;00m\u001b[39m'\u001b[39m)\n\u001b[1;32m---> 37\u001b[0m \u001b[39mprint\u001b[39m(\u001b[39mf\u001b[39m\u001b[39m'\u001b[39m\u001b[39mresult: \u001b[39m\u001b[39m{\u001b[39;00minvocation\u001b[39m.\u001b[39mget_output()\u001b[39m}\u001b[39;00m\u001b[39m'\u001b[39m)\n",
      "File \u001b[1;32mC:\\dev\\enact\\src\\enact\\invocations.py:243\u001b[0m, in \u001b[0;36mInvocation.get_output\u001b[1;34m(self)\u001b[0m\n\u001b[0;32m    241\u001b[0m \u001b[39m\u001b[39m\u001b[39m\"\"\"Returns the output or raise assertion error.\"\"\"\u001b[39;00m\n\u001b[0;32m    242\u001b[0m output \u001b[39m=\u001b[39m \u001b[39mself\u001b[39m\u001b[39m.\u001b[39mresponse()\u001b[39m.\u001b[39moutput\n\u001b[1;32m--> 243\u001b[0m \u001b[39massert\u001b[39;00m output\n\u001b[0;32m    244\u001b[0m \u001b[39mreturn\u001b[39;00m output()\n",
      "\u001b[1;31mAssertionError\u001b[0m: "
     ]
    }
   ],
   "source": [
    "@enact.typed_invokable(enact.NoneResource, enact.Int)\n",
    "@dataclasses.dataclass\n",
    "class AsyncUserRoll(enact.AsyncInvokable):\n",
    "  roll_id: int\n",
    "  async def call(self):\n",
    "    # We provide the roll ID so that we can later distinguish between different\n",
    "    # input requests.\n",
    "    return enact.request_input(enact.Int, for_resource=enact.Int(self.roll_id))\n",
    "\n",
    "@enact.typed_invokable(enact.NoneResource, enact.Int)\n",
    "class CrowdSourcedDiceRoll(enact.AsyncInvokable):\n",
    "  \"\"\"Generates a batch of input requests for each dice roll.\"\"\"\n",
    "  async def call(self):\n",
    "    rolls = [AsyncUserRoll(roll_id=i) for i in range(10)]\n",
    "    # Wait for all rolls to complete\n",
    "    results = await asyncio.gather(*[r() for r in rolls], return_exceptions=True)\n",
    "    for r in results:\n",
    "      if isinstance(r, Exception):\n",
    "        raise r\n",
    "    return enact.Int(sum(results))\n",
    "\n",
    "with store:\n",
    "  invocation = await CrowdSourcedDiceRoll().invoke()\n",
    "  assert isinstance(invocation.get_raised(), enact.InputRequest)\n",
    "\n",
    "  def provide_input(exception_ref):\n",
    "    \"\"\"Resolves an InputRequest exception by providing a dice roll.\"\"\"\n",
    "    input_request = exception_ref()\n",
    "    assert isinstance(input_request, enact.InputRequest)\n",
    "    roll_id = input_request.for_resource()\n",
    "    roll = (roll_id % 7) + 1\n",
    "    print(f'injecting input {roll} for roll {input_request.for_resource()}')\n",
    "    return enact.Int(roll)\n",
    "  \n",
    "  invocation = await invocation.replay_async(exception_override=provide_input)\n",
    "  print(f'result: {invocation.get_raised()}')\n",
    "  print(f'result: {invocation.get_output()}')\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": ".venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.10"
  },
  "orig_nbformat": 4
 },
 "nbformat": 4,
 "nbformat_minor": 2
}