We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Ensure that all dbt invocations have an invocation_id
This is a sub-issue of #4260. 

In order to make sure we can distinguish logs from different runs we should always have an invocation_id. If tracking is off, it isn't created today so we would need to create it. (to add it to logging, add it as a class property on the superclass) (previously considered: preserve process id like it is today, global run_id reported at start, or allow some sort of user markers).

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `core/dbt/lib.py`
Content:
```
1 # TODO: this file is one big TODO
2 import os
3 from dbt.exceptions import RuntimeException
4 from dbt import flags
5 from collections import namedtuple
6 
7 RuntimeArgs = namedtuple(
8     'RuntimeArgs', 'project_dir profiles_dir single_threaded profile_name'
9 )
10 
11 
12 def get_dbt_config(project_dir, single_threaded=False):
13     from dbt.config.runtime import RuntimeConfig
14     import dbt.adapters.factory
15 
16     if os.getenv('DBT_PROFILES_DIR'):
17         profiles_dir = os.getenv('DBT_PROFILES_DIR')
18     else:
19         profiles_dir = os.path.expanduser("~/.dbt")
20 
21     # Construct a phony config
22     config = RuntimeConfig.from_args(RuntimeArgs(
23         project_dir, profiles_dir, single_threaded, 'user'
24     ))
25     # Clear previously registered adapters--
26     # this fixes cacheing behavior on the dbt-server
27     flags.set_from_args('', config)
28     dbt.adapters.factory.reset_adapters()
29     # Load the relevant adapter
30     dbt.adapters.factory.register_adapter(config)
31 
32     return config
33 
34 
35 def get_task_by_type(type):
36     # TODO: we need to tell dbt-server what tasks are available
37     from dbt.task.run import RunTask
38     from dbt.task.list import ListTask
39     from dbt.task.seed import SeedTask
40     from dbt.task.test import TestTask
41     from dbt.task.build import BuildTask
42     from dbt.task.snapshot import SnapshotTask
43     from dbt.task.run_operation import RunOperationTask
44 
45     if type == 'run':
46         return RunTask
47     elif type == 'test':
48         return TestTask
49     elif type == 'list':
50         return ListTask
51     elif type == 'seed':
52         return SeedTask
53     elif type == 'build':
54         return BuildTask
55     elif type == 'snapshot':
56         return SnapshotTask
57     elif type == 'run_operation':
58         return RunOperationTask
59 
60     raise RuntimeException('not a valid task')
61 
62 
63 def create_task(type, args, manifest, config):
64     task = get_task_by_type(type)
65 
66     def no_op(*args, **kwargs):
67         pass
68 
69     # TODO: yuck, let's rethink tasks a little
70     task = task(args, config)
71 
72     # Wow! We can monkeypatch taskCls.load_manifest to return _our_ manifest
73     task.load_manifest = no_op
74     task.manifest = manifest
75     return task
76 
77 
78 def _get_operation_node(manifest, project_path, sql):
79     from dbt.parser.manifest import process_node
80     from dbt.parser.sql import SqlBlockParser
81     import dbt.adapters.factory
82 
83     config = get_dbt_config(project_path)
84     block_parser = SqlBlockParser(
85         project=config,
86         manifest=manifest,
87         root_project=config,
88     )
89 
90     adapter = dbt.adapters.factory.get_adapter(config)
91     # TODO : This needs a real name?
92     sql_node = block_parser.parse_remote(sql, 'name')
93     process_node(config, manifest, sql_node)
94     return config, sql_node, adapter
95 
96 
97 def compile_sql(manifest, project_path, sql):
98     from dbt.task.sql import SqlCompileRunner
99 
100     config, node, adapter = _get_operation_node(manifest, project_path, sql)
101     runner = SqlCompileRunner(config, adapter, node, 1, 1)
102     return runner.safe_run(manifest)
103 
104 
105 def execute_sql(manifest, project_path, sql):
106     from dbt.task.sql import SqlExecuteRunner
107 
108     config, node, adapter = _get_operation_node(manifest, project_path, sql)
109     runner = SqlExecuteRunner(config, adapter, node, 1, 1)
110     # TODO: use same interface for runner
111     return runner.safe_run(manifest)
112 
113 
114 def parse_to_manifest(config):
115     from dbt.parser.manifest import ManifestLoader
116 
117     return ManifestLoader.get_full_manifest(config)
118 
119 
120 def deserialize_manifest(manifest_msgpack):
121     from dbt.contracts.graph.manifest import Manifest
122 
123     return Manifest.from_msgpack(manifest_msgpack)
124 
125 
126 def serialize_manifest(manifest):
127     # TODO: what should this take as an arg?
128     return manifest.to_msgpack()
129
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/core/dbt/lib.py b/core/dbt/lib.py
--- a/core/dbt/lib.py
+++ b/core/dbt/lib.py
@@ -12,6 +12,7 @@
 def get_dbt_config(project_dir, single_threaded=False):
     from dbt.config.runtime import RuntimeConfig
     import dbt.adapters.factory
+    import dbt.events.functions
 
     if os.getenv('DBT_PROFILES_DIR'):
         profiles_dir = os.getenv('DBT_PROFILES_DIR')
@@ -28,6 +29,8 @@
     dbt.adapters.factory.reset_adapters()
     # Load the relevant adapter
     dbt.adapters.factory.register_adapter(config)
+    # Set invocation id
+    dbt.events.functions.set_invocation_id()
 
     return config

{"golden_diff": "diff --git a/core/dbt/lib.py b/core/dbt/lib.py\n--- a/core/dbt/lib.py\n+++ b/core/dbt/lib.py\n@@ -12,6 +12,7 @@\n def get_dbt_config(project_dir, single_threaded=False):\n     from dbt.config.runtime import RuntimeConfig\n     import dbt.adapters.factory\n+    import dbt.events.functions\n \n     if os.getenv('DBT_PROFILES_DIR'):\n         profiles_dir = os.getenv('DBT_PROFILES_DIR')\n@@ -28,6 +29,8 @@\n     dbt.adapters.factory.reset_adapters()\n     # Load the relevant adapter\n     dbt.adapters.factory.register_adapter(config)\n+    # Set invocation id\n+    dbt.events.functions.set_invocation_id()\n \n     return config\n", "issue": "Ensure that all dbt invocations have an invocation_id\nThis is a sub-issue of #4260. \n\nIn order to make sure we can distinguish logs from different runs we should always have an invocation_id. If tracking is off, it isn't created today so we would need to create it. (to add it to logging, add it as a class property on the superclass) (previously considered: preserve process id like it is today, global run_id reported at start, or allow some sort of user markers).\n", "before_files": [{"content": "# TODO: this file is one big TODO\nimport os\nfrom dbt.exceptions import RuntimeException\nfrom dbt import flags\nfrom collections import namedtuple\n\nRuntimeArgs = namedtuple(\n    'RuntimeArgs', 'project_dir profiles_dir single_threaded profile_name'\n)\n\n\ndef get_dbt_config(project_dir, single_threaded=False):\n    from dbt.config.runtime import RuntimeConfig\n    import dbt.adapters.factory\n\n    if os.getenv('DBT_PROFILES_DIR'):\n        profiles_dir = os.getenv('DBT_PROFILES_DIR')\n    else:\n        profiles_dir = os.path.expanduser(\"~/.dbt\")\n\n    # Construct a phony config\n    config = RuntimeConfig.from_args(RuntimeArgs(\n        project_dir, profiles_dir, single_threaded, 'user'\n    ))\n    # Clear previously registered adapters--\n    # this fixes cacheing behavior on the dbt-server\n    flags.set_from_args('', config)\n    dbt.adapters.factory.reset_adapters()\n    # Load the relevant adapter\n    dbt.adapters.factory.register_adapter(config)\n\n    return config\n\n\ndef get_task_by_type(type):\n    # TODO: we need to tell dbt-server what tasks are available\n    from dbt.task.run import RunTask\n    from dbt.task.list import ListTask\n    from dbt.task.seed import SeedTask\n    from dbt.task.test import TestTask\n    from dbt.task.build import BuildTask\n    from dbt.task.snapshot import SnapshotTask\n    from dbt.task.run_operation import RunOperationTask\n\n    if type == 'run':\n        return RunTask\n    elif type == 'test':\n        return TestTask\n    elif type == 'list':\n        return ListTask\n    elif type == 'seed':\n        return SeedTask\n    elif type == 'build':\n        return BuildTask\n    elif type == 'snapshot':\n        return SnapshotTask\n    elif type == 'run_operation':\n        return RunOperationTask\n\n    raise RuntimeException('not a valid task')\n\n\ndef create_task(type, args, manifest, config):\n    task = get_task_by_type(type)\n\n    def no_op(*args, **kwargs):\n        pass\n\n    # TODO: yuck, let's rethink tasks a little\n    task = task(args, config)\n\n    # Wow! We can monkeypatch taskCls.load_manifest to return _our_ manifest\n    task.load_manifest = no_op\n    task.manifest = manifest\n    return task\n\n\ndef _get_operation_node(manifest, project_path, sql):\n    from dbt.parser.manifest import process_node\n    from dbt.parser.sql import SqlBlockParser\n    import dbt.adapters.factory\n\n    config = get_dbt_config(project_path)\n    block_parser = SqlBlockParser(\n        project=config,\n        manifest=manifest,\n        root_project=config,\n    )\n\n    adapter = dbt.adapters.factory.get_adapter(config)\n    # TODO : This needs a real name?\n    sql_node = block_parser.parse_remote(sql, 'name')\n    process_node(config, manifest, sql_node)\n    return config, sql_node, adapter\n\n\ndef compile_sql(manifest, project_path, sql):\n    from dbt.task.sql import SqlCompileRunner\n\n    config, node, adapter = _get_operation_node(manifest, project_path, sql)\n    runner = SqlCompileRunner(config, adapter, node, 1, 1)\n    return runner.safe_run(manifest)\n\n\ndef execute_sql(manifest, project_path, sql):\n    from dbt.task.sql import SqlExecuteRunner\n\n    config, node, adapter = _get_operation_node(manifest, project_path, sql)\n    runner = SqlExecuteRunner(config, adapter, node, 1, 1)\n    # TODO: use same interface for runner\n    return runner.safe_run(manifest)\n\n\ndef parse_to_manifest(config):\n    from dbt.parser.manifest import ManifestLoader\n\n    return ManifestLoader.get_full_manifest(config)\n\n\ndef deserialize_manifest(manifest_msgpack):\n    from dbt.contracts.graph.manifest import Manifest\n\n    return Manifest.from_msgpack(manifest_msgpack)\n\n\ndef serialize_manifest(manifest):\n    # TODO: what should this take as an arg?\n    return manifest.to_msgpack()\n", "path": "core/dbt/lib.py"}], "after_files": [{"content": "# TODO: this file is one big TODO\nimport os\nfrom dbt.exceptions import RuntimeException\nfrom dbt import flags\nfrom collections import namedtuple\n\nRuntimeArgs = namedtuple(\n    'RuntimeArgs', 'project_dir profiles_dir single_threaded profile_name'\n)\n\n\ndef get_dbt_config(project_dir, single_threaded=False):\n    from dbt.config.runtime import RuntimeConfig\n    import dbt.adapters.factory\n    import dbt.events.functions\n\n    if os.getenv('DBT_PROFILES_DIR'):\n        profiles_dir = os.getenv('DBT_PROFILES_DIR')\n    else:\n        profiles_dir = os.path.expanduser(\"~/.dbt\")\n\n    # Construct a phony config\n    config = RuntimeConfig.from_args(RuntimeArgs(\n        project_dir, profiles_dir, single_threaded, 'user'\n    ))\n    # Clear previously registered adapters--\n    # this fixes cacheing behavior on the dbt-server\n    flags.set_from_args('', config)\n    dbt.adapters.factory.reset_adapters()\n    # Load the relevant adapter\n    dbt.adapters.factory.register_adapter(config)\n    # Set invocation id\n    dbt.events.functions.set_invocation_id()\n\n    return config\n\n\ndef get_task_by_type(type):\n    # TODO: we need to tell dbt-server what tasks are available\n    from dbt.task.run import RunTask\n    from dbt.task.list import ListTask\n    from dbt.task.seed import SeedTask\n    from dbt.task.test import TestTask\n    from dbt.task.build import BuildTask\n    from dbt.task.snapshot import SnapshotTask\n    from dbt.task.run_operation import RunOperationTask\n\n    if type == 'run':\n        return RunTask\n    elif type == 'test':\n        return TestTask\n    elif type == 'list':\n        return ListTask\n    elif type == 'seed':\n        return SeedTask\n    elif type == 'build':\n        return BuildTask\n    elif type == 'snapshot':\n        return SnapshotTask\n    elif type == 'run_operation':\n        return RunOperationTask\n\n    raise RuntimeException('not a valid task')\n\n\ndef create_task(type, args, manifest, config):\n    task = get_task_by_type(type)\n\n    def no_op(*args, **kwargs):\n        pass\n\n    # TODO: yuck, let's rethink tasks a little\n    task = task(args, config)\n\n    # Wow! We can monkeypatch taskCls.load_manifest to return _our_ manifest\n    task.load_manifest = no_op\n    task.manifest = manifest\n    return task\n\n\ndef _get_operation_node(manifest, project_path, sql):\n    from dbt.parser.manifest import process_node\n    from dbt.parser.sql import SqlBlockParser\n    import dbt.adapters.factory\n\n    config = get_dbt_config(project_path)\n    block_parser = SqlBlockParser(\n        project=config,\n        manifest=manifest,\n        root_project=config,\n    )\n\n    adapter = dbt.adapters.factory.get_adapter(config)\n    # TODO : This needs a real name?\n    sql_node = block_parser.parse_remote(sql, 'name')\n    process_node(config, manifest, sql_node)\n    return config, sql_node, adapter\n\n\ndef compile_sql(manifest, project_path, sql):\n    from dbt.task.sql import SqlCompileRunner\n\n    config, node, adapter = _get_operation_node(manifest, project_path, sql)\n    runner = SqlCompileRunner(config, adapter, node, 1, 1)\n    return runner.safe_run(manifest)\n\n\ndef execute_sql(manifest, project_path, sql):\n    from dbt.task.sql import SqlExecuteRunner\n\n    config, node, adapter = _get_operation_node(manifest, project_path, sql)\n    runner = SqlExecuteRunner(config, adapter, node, 1, 1)\n    # TODO: use same interface for runner\n    return runner.safe_run(manifest)\n\n\ndef parse_to_manifest(config):\n    from dbt.parser.manifest import ManifestLoader\n\n    return ManifestLoader.get_full_manifest(config)\n\n\ndef deserialize_manifest(manifest_msgpack):\n    from dbt.contracts.graph.manifest import Manifest\n\n    return Manifest.from_msgpack(manifest_msgpack)\n\n\ndef serialize_manifest(manifest):\n    # TODO: what should this take as an arg?\n    return manifest.to_msgpack()\n", "path": "core/dbt/lib.py"}]}